Escape Telegram Markdown characters in error messages and enhance test coverage.

- Add `escape_markdown` utility function to sanitize dynamic content in Telegram messages.
- Update `ErrorReporter` to escape special characters in error messages, URLs, user agents, and additional context fields.
- Refactor and expand test cases for `escape_markdown` function under various scenarios, including real-world error messages.
- Adjust test mocks in `kis.py` to reflect changes in method naming for better clarity (`get_holdings` -> `get_holdings_by_user`).

Author: robin-watcha

app/monitoring/error_reporter.py

@@ -10,6 +10,7 @@
 
 import hashlib
 import logging
+import re
 import traceback
 from typing import Dict, Optional
 
@@ -22,6 +23,17 @@
 logger = logging.getLogger(__name__)
 
 
+def escape_markdown(text: str) -> str:
+    """Escape Telegram Markdown special characters.
+
+    Telegram Markdown v1 reserves these characters: _ * ` [
+    """
+    # 백틱(`) 안의 텍스트는 그대로 두고, 나머지만 이스케이프
+    # 간단하게 모든 특수문자 이스케이프
+    escape_chars = r'_*`['
+    return re.sub(f'([{re.escape(escape_chars)}])', r'\\\1', text)
+
+
 class ErrorReporter:
     """
     Singleton error reporter with Telegram integration and Redis-based deduplication.
@@ -177,13 +189,16 @@ def _format_error_message(
         """
         timestamp = format_datetime()
 
+        # Escape Markdown special characters in dynamic content
+        safe_error_message = escape_markdown(error_message)
+
         # Build message parts
         parts = [
             "🚨 *Error Alert*",
             f"🕒 {timestamp}",
             "",
             f"*Type:* `{error_type}`",
-            f"*Message:* {error_message}",
+            f"*Message:* {safe_error_message}",
         ]
 
         # Add request info if available
@@ -193,26 +208,29 @@ def _format_error_message(
             if "method" in request_info:
                 parts.append(f"  • Method: `{request_info['method']}`")
             if "url" in request_info:
-                parts.append(f"  • URL: `{request_info['url']}`")
+                safe_url = escape_markdown(str(request_info['url']))
+                parts.append(f"  • URL: {safe_url}")
             if "client" in request_info:
                 parts.append(f"  • Client: `{request_info['client']}`")
             if "user_agent" in request_info:
                 user_agent = request_info["user_agent"][:100]  # Truncate
-                parts.append(f"  • User-Agent: `{user_agent}`")
+                safe_ua = escape_markdown(user_agent)
+                parts.append(f"  • User-Agent: {safe_ua}")
 
         # Add additional context if available
         if additional_context:
             parts.append("")
             parts.append("*Additional Context:*")
             for key, value in additional_context.items():
                 # Format specific keys nicely
+                safe_value = escape_markdown(str(value))
                 if key == "request_id":
                     parts.append(f"  • Request ID: `{value}`")
                 elif key == "duration_ms":
                     parts.append(f"  • Duration: `{value:.2f}ms`")
                 else:
-                    # Generic key-value pair
-                    parts.append(f"  • {key}: `{value}`")
+                    # Generic key-value pair - escape value
+                    parts.append(f"  • {key}: {safe_value}")
 
         # Add stack trace (truncated if too long)
         parts.append("")

app/tasks/kis.py

@@ -438,7 +438,7 @@ async def _run() -> dict:
                 manual_service = ManualHoldingsService(db)
                 # USER_ID는 현재 1로 고정 (추후 다중 사용자 지원 시 변경 필요)
                 user_id = 1
-                manual_holdings = await manual_service.get_holdings(user_id=user_id, market_type=MarketType.KR)
+                manual_holdings = await manual_service.get_holdings_by_user(user_id=user_id, market_type=MarketType.KR)
 
             # 3. 수동 잔고 종목을 한투 형식으로 변환하여 병합
             for holding in manual_holdings:

tests/test_kis_tasks.py

@@ -42,7 +42,7 @@ class MockManualService:
         def __init__(self, db):
             pass
 
-        async def get_holdings(self, user_id, market_type):
+        async def get_holdings_by_user(self, user_id, market_type):
             return []  # No manual holdings
 
     buy_calls = []
@@ -194,7 +194,7 @@ class MockManualService:
         def __init__(self, db):
             pass
 
-        async def get_holdings(self, user_id, market_type):
+        async def get_holdings_by_user(self, user_id, market_type):
             return []  # No manual holdings
 
     with patch('app.core.db.AsyncSessionLocal') as mock_session_cls, \
@@ -281,7 +281,7 @@ class MockManualService:
         def __init__(self, db):
             pass
 
-        async def get_holdings(self, user_id, market_type):
+        async def get_holdings_by_user(self, user_id, market_type):
             return []
 
     sell_calls = []
@@ -369,7 +369,7 @@ class MockManualService:
         def __init__(self, db):
             pass
 
-        async def get_holdings(self, user_id, market_type):
+        async def get_holdings_by_user(self, user_id, market_type):
             return []
 
     async def fake_buy(*_, **__):
@@ -462,7 +462,7 @@ class MockManualService:
         def __init__(self, db):
             pass
 
-        async def get_holdings(self, user_id, market_type):
+        async def get_holdings_by_user(self, user_id, market_type):
             return []
 
     sell_calls: List[Dict[str, Any]] = []
@@ -635,7 +635,7 @@ class MockManualService:
             def __init__(self, db):
                 pass
 
-            async def get_holdings(self, user_id, market_type):
+            async def get_holdings_by_user(self, user_id, market_type):
                 return []
 
         error_reports = []
@@ -723,7 +723,7 @@ class MockManualService:
             def __init__(self, db):
                 pass
 
-            async def get_holdings(self, user_id, market_type):
+            async def get_holdings_by_user(self, user_id, market_type):
                 return []
 
         error_reports = []
@@ -1247,7 +1247,7 @@ class MockManualService:
             def __init__(self, db):
                 pass
 
-            async def get_holdings(self, user_id, market_type):
+            async def get_holdings_by_user(self, user_id, market_type):
                 return []
 
         async def fake_buy(*_, **__):
@@ -1339,7 +1339,7 @@ class MockManualService:
             def __init__(self, db):
                 pass
 
-            async def get_holdings(self, user_id, market_type):
+            async def get_holdings_by_user(self, user_id, market_type):
                 return []
 
         async def fake_buy(*_, **__):
@@ -1432,7 +1432,7 @@ class MockManualService:
             def __init__(self, db):
                 pass
 
-            async def get_holdings(self, user_id, market_type):
+            async def get_holdings_by_user(self, user_id, market_type):
                 return []
 
         async def fake_buy(*_, **__):

tests/test_monitoring.py

@@ -8,7 +8,7 @@
 from fastapi import HTTPException
 
 from app.middleware.monitoring import MonitoringMiddleware
-from app.monitoring.error_reporter import ErrorReporter
+from app.monitoring.error_reporter import ErrorReporter, escape_markdown
 
 
 @pytest.fixture
@@ -132,3 +132,74 @@ def __str__(self):
     assert reporter.last_context["status_code"] == 500
     assert reporter.last_context["request_id"] == "req-1"
     assert isinstance(reporter.last_context["duration_ms"], float)
+
+
+class TestEscapeMarkdown:
+    """escape_markdown 함수 테스트."""
+
+    def test_escape_underscore(self):
+        """언더스코어(_)가 이스케이프되어야 함."""
+        text = "process_kis_domestic_sell_orders"
+        result = escape_markdown(text)
+        assert result == r"process\_kis\_domestic\_sell\_orders"
+
+    def test_escape_asterisk(self):
+        """별표(*)가 이스케이프되어야 함."""
+        text = "**bold** and *italic*"
+        result = escape_markdown(text)
+        assert r"\*\*bold\*\*" in result
+        assert r"\*italic\*" in result
+
+    def test_escape_backtick(self):
+        """백틱(`)이 이스케이프되어야 함."""
+        text = "use `code` here"
+        result = escape_markdown(text)
+        assert r"\`code\`" in result
+
+    def test_escape_square_bracket(self):
+        """대괄호([)가 이스케이프되어야 함."""
+        text = "see [link] here"
+        result = escape_markdown(text)
+        assert r"\[link]" in result
+
+    def test_no_escape_for_normal_text(self):
+        """특수문자가 없는 일반 텍스트는 변경되지 않아야 함."""
+        text = "APBK0400 주문 가능한 수량을 초과했습니다."
+        result = escape_markdown(text)
+        assert result == text
+
+    def test_complex_error_message(self):
+        """실제 에러 메시지 시나리오."""
+        text = "File '/app/services/kis_trading_service.py', line 343"
+        result = escape_markdown(text)
+        # 언더스코어만 이스케이프됨
+        assert r"kis\_trading\_service" in result
+
+
+class TestFormatErrorMessageWithEscape:
+    """에러 메시지 포맷팅 시 이스케이프 테스트."""
+
+    def test_format_message_escapes_error_message(self, error_reporter):
+        """에러 메시지의 특수문자가 이스케이프되어야 함."""
+        message = error_reporter._format_error_message(
+            error_type="RuntimeError",
+            error_message="Error in func_name with _underscore",
+            stack_trace="simple trace",
+        )
+        # 에러 메시지 부분에서 언더스코어가 이스케이프되어야 함
+        assert r"func\_name" in message
+        assert r"\_underscore" in message
+
+    def test_format_message_escapes_additional_context(self, error_reporter):
+        """추가 컨텍스트의 특수문자가 이스케이프되어야 함."""
+        message = error_reporter._format_error_message(
+            error_type="RuntimeError",
+            error_message="Error",
+            stack_trace="trace",
+            additional_context={
+                "task_name": "kis.run_per_domestic_stock_automation",
+                "stock": "삼성전자우 (005935)"
+            }
+        )
+        # task_name의 언더스코어가 이스케이프되어야 함
+        assert r"run\_per\_domestic\_stock\_automation" in message