diff --git a/apps/api/src/plugins/mcp/providers/ssh_provider.py b/apps/api/src/plugins/mcp/providers/ssh_provider.py index 3dac99f4..9ee04e19 100644 --- a/apps/api/src/plugins/mcp/providers/ssh_provider.py +++ b/apps/api/src/plugins/mcp/providers/ssh_provider.py @@ -41,6 +41,7 @@ SSH 連線: @see docs/superpowers/specs/2026-04-10-infra-rebuild-sprint-abc-design.md §MCP-2a """ +import logging import re import uuid from datetime import UTC, datetime @@ -51,6 +52,7 @@ import structlog from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult logger = structlog.get_logger(__name__) +_asyncssh_logger_configured = False # ============================================================================= # 安全常數 @@ -128,6 +130,22 @@ def _normalize_ssh_host(value: str) -> str: return maybe_host return host + +def _quiet_asyncssh_info_logs() -> None: + """Keep third-party asyncssh INFO logs from breaking stdlib %-format logging. + + Some target SSH servers send exit status as a string. AsyncSSH then emits an + INFO log with ``%d`` and that string argument before our code sees the + result, which produces noisy ``TypeError: %d format`` tracebacks. The tool + result itself is still available, so production should keep asyncssh at + WARNING and rely on our structured MCP audit logs. + """ + global _asyncssh_logger_configured + if _asyncssh_logger_configured: + return + logging.getLogger("asyncssh").setLevel(logging.WARNING) + _asyncssh_logger_configured = True + # 群組 A(只讀) GROUP_A_TOOLS = { "ssh_diagnose", @@ -630,6 +648,8 @@ class SSHProvider(MCPToolProvider): "Add 'asyncssh' to pyproject.toml dependencies." ) from None + _quiet_asyncssh_info_logs() + import os if not os.path.exists(SSH_KEY_PATH): raise RuntimeError( diff --git a/apps/api/tests/test_ssh_provider_tools.py b/apps/api/tests/test_ssh_provider_tools.py index 55f380df..e9d14e13 100644 --- a/apps/api/tests/test_ssh_provider_tools.py +++ b/apps/api/tests/test_ssh_provider_tools.py @@ -1,6 +1,13 @@ +import logging + import pytest -from src.plugins.mcp.providers.ssh_provider import SSHProvider, _normalize_ssh_host +import src.plugins.mcp.providers.ssh_provider as ssh_provider_module +from src.plugins.mcp.providers.ssh_provider import ( + SSHProvider, + _normalize_ssh_host, + _quiet_asyncssh_info_logs, +) @pytest.mark.asyncio @@ -34,6 +41,22 @@ def test_normalize_ssh_host_strips_exporter_ports_and_users(raw, expected): assert _normalize_ssh_host(raw) == expected +def test_quiet_asyncssh_info_logs_sets_asyncssh_to_warning(monkeypatch): + monkeypatch.setattr(ssh_provider_module, "_asyncssh_logger_configured", False) + asyncssh_logger = logging.getLogger("asyncssh") + previous_level = asyncssh_logger.level + asyncssh_logger.setLevel(logging.INFO) + + try: + _quiet_asyncssh_info_logs() + + assert asyncssh_logger.level == logging.WARNING + assert ssh_provider_module._asyncssh_logger_configured is True + finally: + asyncssh_logger.setLevel(previous_level) + ssh_provider_module._asyncssh_logger_configured = False + + @pytest.mark.asyncio async def test_ssh_execute_normalizes_host_before_allowed_check(monkeypatch): provider = SSHProvider()