fix(mcp): audit approved ssh execution path

apps/api/src/services/approval_execution.py

@@ -685,10 +685,36 @@ class ApprovalExecutionService:
             )
 
         # 呼叫 SSH MCP Provider
+        # 2026-05-06 Codex: approved execution 是高風險「實際執行」路徑。
+        # 在 AwoooP MCP Gateway 完全接管前，至少必須經過 AuditedMCPToolProvider
+        # 寫入 durable mcp_audit_log，並標記這仍是 legacy direct provider path。
         from src.plugins.mcp.providers.ssh_provider import SSHProvider
-        provider = SSHProvider()
+        from src.plugins.mcp.registry import AuditedMCPToolProvider
+
+        provider = AuditedMCPToolProvider(SSHProvider())
+        params_with_audit = {
+            **params,
+            "_mcp_audit": {
+                "session_id": f"approval:{approval.id}",
+                "incident_id": approval.incident_id,
+                "agent_role": "approval_executor",
+                "flywheel_node": "execute",
+                "gateway_path": "legacy_direct_provider",
+            },
+        }
         try:
-            mcp_result = await provider.execute(tool_name=tool_name, parameters=params)
+            logger.warning(
+                "mcp_gateway_legacy_direct_provider_path",
+                approval_id=str(approval.id),
+                incident_id=approval.incident_id,
+                tool=tool_name,
+                host=host,
+                reason="awooop_gateway_not_enforced_for_legacy_approval_execution",
+            )
+            mcp_result = await provider.execute(
+                tool_name=tool_name,
+                parameters=params_with_audit,
+            )
             duration_ms = int((time.time() - start) * 1000)
             success = bool(mcp_result.success)
             return ExecutionResult(

apps/api/tests/test_approval_execution_mcp_audit.py

@@ -0,0 +1,71 @@
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from src.models.approval import ApprovalRequest, RiskLevel
+from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
+from src.services.approval_execution import ApprovalExecutionService
+
+
+class FakeSSHProvider(MCPToolProvider):
+    name = "ssh"
+    enabled = True
+    seen_parameters: dict[str, Any] | None = None
+
+    async def list_tools(self) -> list[MCPTool]:
+        return []
+
+    async def execute(self, tool_name: str, parameters: dict) -> MCPToolResult:
+        self.seen_parameters = dict(parameters)
+        return MCPToolResult(
+            success=True,
+            output={"tool": tool_name, "ok": True},
+            execution_id="fake-ssh-exec",
+        )
+
+    async def health_check(self) -> bool:
+        return True
+
+
+@pytest.mark.asyncio
+async def test_ssh_approval_execution_uses_audited_provider(monkeypatch: pytest.MonkeyPatch) -> None:
+    fake_provider = FakeSSHProvider()
+    audit_calls: list[dict[str, Any]] = []
+
+    class ProviderFactory:
+        def __call__(self) -> FakeSSHProvider:
+            return fake_provider
+
+    async def fake_record_mcp_call(**kwargs: Any) -> None:
+        audit_calls.append(kwargs)
+
+    monkeypatch.setattr("src.plugins.mcp.providers.ssh_provider.SSHProvider", ProviderFactory())
+    monkeypatch.setattr("src.services.mcp_audit_service.record_mcp_call", fake_record_mcp_call)
+
+    approval = ApprovalRequest(
+        action="docker restart sentry-worker",
+        description="測試 SSH approved execution audit",
+        risk_level=RiskLevel.LOW,
+        requested_by="test",
+        required_signatures=0,
+        incident_id="INC-TEST-AUDIT",
+    )
+
+    result = await ApprovalExecutionService()._execute_ssh_host_action(
+        approval=approval,
+        host="192.168.0.110",
+    )
+
+    assert result.success is True
+    assert fake_provider.seen_parameters is not None
+    assert "_mcp_audit" not in fake_provider.seen_parameters
+    assert audit_calls
+    audit = audit_calls[0]
+    assert audit["mcp_server"] == "ssh"
+    assert audit["tool_name"] == "ssh_docker_restart"
+    assert audit["incident_id"] == "INC-TEST-AUDIT"
+    assert audit["agent_role"] == "approval_executor"
+    assert audit["flywheel_node"] == "execute"
+    assert audit["input_params"]["_mcp_audit"]["gateway_path"] == "legacy_direct_provider"