452 lines
14 KiB
Python
452 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock
|
|
from uuid import UUID
|
|
|
|
import pytest
|
|
|
|
from src.models.approval import ApprovalRequest, RiskLevel
|
|
from src.plugins.mcp.gateway import GateApprovalError
|
|
from src.plugins.mcp.interfaces import MCPToolResult
|
|
from src.services.approval_execution import ApprovalExecutionService
|
|
from src.services.executor import ExecutionResult, OperationType
|
|
|
|
|
|
class FakeRedis:
|
|
def __init__(self) -> None:
|
|
self.set_calls: list[tuple[str, str, int | None]] = []
|
|
|
|
async def set(self, key: str, value: str, ex: int | None = None) -> None:
|
|
self.set_calls.append((key, value, ex))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_systemctl_approval_queues_ansible_without_ssh_gateway(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
queue = AsyncMock(
|
|
return_value={
|
|
"status": "controlled_check_mode_queued",
|
|
"queued": True,
|
|
"automation_run_id": "run-systemd-approval",
|
|
"single_writer_executor": "awoooi-ansible-executor-broker",
|
|
"runtime_write_performed": False,
|
|
"repair_executed": False,
|
|
"repair_verified": False,
|
|
}
|
|
)
|
|
gateway = AsyncMock(
|
|
side_effect=AssertionError("systemctl must not use the SSH gateway")
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.host_ansible_controlled_executor."
|
|
"queue_host_ansible_controlled_action_for_incident_id",
|
|
queue,
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_execute_ssh_tool_via_gateway",
|
|
gateway,
|
|
)
|
|
approval = ApprovalRequest(
|
|
action="ssh 192.168.0.110 'systemctl restart node_exporter'",
|
|
description="typed Host/systemd approval",
|
|
risk_level=RiskLevel.MEDIUM,
|
|
requested_by="test",
|
|
required_signatures=0,
|
|
incident_id="INC-NODE-EXPORTER-110",
|
|
)
|
|
|
|
result = await ApprovalExecutionService()._execute_ssh_host_action(
|
|
approval=approval,
|
|
host="192.168.0.110",
|
|
)
|
|
|
|
assert result.success is True
|
|
assert result.k8s_response is not None
|
|
assert result.k8s_response["controlled_execution_queued"] is True
|
|
assert result.k8s_response["runtime_write_performed"] is False
|
|
assert result.k8s_response["repair_executed"] is False
|
|
queue.assert_awaited_once()
|
|
gateway.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"action",
|
|
[
|
|
"ssh://root@192.168.0.110/systemctl%20start%20node_exporter",
|
|
"ssh 192.168.0.110 'systemctl restart node_exporter || true'",
|
|
(
|
|
"ssh 192.168.0.110 'docker restart exporter "
|
|
"|| systemctl restart node_exporter'"
|
|
),
|
|
],
|
|
)
|
|
async def test_unbounded_systemd_approval_never_reaches_any_executor(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
action: str,
|
|
) -> None:
|
|
queue = AsyncMock(
|
|
side_effect=AssertionError("invalid systemd action must not be queued")
|
|
)
|
|
gateway = AsyncMock(
|
|
side_effect=AssertionError("invalid systemd action must not use SSH MCP")
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.host_ansible_controlled_executor."
|
|
"queue_host_ansible_controlled_action_for_incident_id",
|
|
queue,
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_execute_ssh_tool_via_gateway",
|
|
gateway,
|
|
)
|
|
approval = ApprovalRequest(
|
|
action=action,
|
|
description="unbounded Host/systemd approval",
|
|
risk_level=RiskLevel.MEDIUM,
|
|
requested_by="test",
|
|
required_signatures=0,
|
|
incident_id="INC-NODE-EXPORTER-110",
|
|
)
|
|
|
|
result = await ApprovalExecutionService()._execute_ssh_host_action(
|
|
approval=approval,
|
|
host="192.168.0.110",
|
|
)
|
|
|
|
assert result.success is False
|
|
assert "unbounded_or_unallowlisted" in str(result.error)
|
|
queue.assert_not_awaited()
|
|
gateway.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_queued_host_approval_does_not_close_or_learn_as_repair(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
approval = ApprovalRequest(
|
|
action="ssh 192.168.0.110 'systemctl restart node_exporter'",
|
|
description="typed Host/systemd approval",
|
|
risk_level=RiskLevel.MEDIUM,
|
|
requested_by="operator",
|
|
required_signatures=0,
|
|
incident_id="INC-NODE-EXPORTER-110",
|
|
)
|
|
handoff = {
|
|
"status": "controlled_check_mode_queued",
|
|
"queued": True,
|
|
"automation_run_id": "run-systemd-approval",
|
|
"single_writer_executor": "awoooi-ansible-executor-broker",
|
|
"runtime_write_performed": False,
|
|
"repair_executed": False,
|
|
"repair_verified": False,
|
|
}
|
|
execute_ssh = AsyncMock(
|
|
return_value=ExecutionResult(
|
|
success=True,
|
|
message="host_ansible_controlled_executor queued",
|
|
operation_type=OperationType.SSH_HOST,
|
|
target_resource="192.168.0.110",
|
|
namespace="host",
|
|
duration_ms=1,
|
|
k8s_response={
|
|
"controlled_execution_queued": True,
|
|
"runtime_write_performed": False,
|
|
"controlled_handoff": handoff,
|
|
},
|
|
)
|
|
)
|
|
update_status = AsyncMock()
|
|
record_pending = AsyncMock(return_value=True)
|
|
timeline = SimpleNamespace(add_event=AsyncMock())
|
|
log_completed = AsyncMock()
|
|
alert_completed = AsyncMock()
|
|
push_result = AsyncMock()
|
|
|
|
monkeypatch.setattr(
|
|
"src.services.approval_execution.get_approval_service",
|
|
lambda: SimpleNamespace(
|
|
update_execution_status=update_status,
|
|
record_controlled_handoff_pending=record_pending,
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.approval_execution.get_timeline_service",
|
|
lambda: timeline,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.approval_execution.get_executor",
|
|
lambda: object(),
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.resource_resolver.get_resource_resolver",
|
|
lambda: SimpleNamespace(
|
|
resolve=AsyncMock(
|
|
return_value=SimpleNamespace(
|
|
success=False,
|
|
resource_name=None,
|
|
candidates=[],
|
|
)
|
|
)
|
|
),
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_execute_ssh_host_action",
|
|
execute_ssh,
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_log_aol_started",
|
|
AsyncMock(return_value="aol-op-id"),
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_log_alert_execution_started",
|
|
AsyncMock(),
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_log_aol_completed",
|
|
log_completed,
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_log_alert_execution_completed",
|
|
alert_completed,
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_push_execution_result_to_alert",
|
|
push_result,
|
|
)
|
|
learning = AsyncMock(
|
|
side_effect=AssertionError("queued handoff must not trigger learning")
|
|
)
|
|
resolve = AsyncMock(
|
|
side_effect=AssertionError("queued handoff must not resolve incident")
|
|
)
|
|
monkeypatch.setattr(
|
|
ApprovalExecutionService,
|
|
"_trigger_learning",
|
|
learning,
|
|
)
|
|
monkeypatch.setattr(
|
|
"src.services.incident_service.get_incident_service",
|
|
lambda: SimpleNamespace(resolve_incident=resolve),
|
|
)
|
|
|
|
result = await ApprovalExecutionService().execute_approved_action(approval)
|
|
|
|
assert result is False
|
|
update_status.assert_not_awaited()
|
|
record_pending.assert_awaited_once_with(
|
|
approval.id,
|
|
execution_kind="host_ansible_check_mode_queued",
|
|
receipt=handoff,
|
|
)
|
|
assert timeline.add_event.await_args.kwargs["status"] == "pending"
|
|
assert "尚未修復" in timeline.add_event.await_args.kwargs["title"]
|
|
alert_completed.assert_not_awaited()
|
|
assert log_completed.await_args.kwargs["status"] == "dry_run"
|
|
assert log_completed.await_args.kwargs["output"]["incident_closure_allowed"] is False
|
|
push_result.assert_awaited_once_with(
|
|
approval,
|
|
success=False,
|
|
error=None,
|
|
execution_kind="host_ansible_check_mode_queued",
|
|
repair_executed=False,
|
|
repair_attempted=False,
|
|
)
|
|
learning.assert_not_awaited()
|
|
resolve.assert_not_awaited()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ssh_approval_execution_projects_approval_into_gateway(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_redis = FakeRedis()
|
|
gateway_calls: list[dict[str, Any]] = []
|
|
db_context_projects: list[str | None] = []
|
|
fake_db = object()
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(project_id: str | None = None):
|
|
db_context_projects.append(project_id)
|
|
yield fake_db
|
|
|
|
class FakeGateway:
|
|
def __init__(self, db: object) -> None:
|
|
self.db = db
|
|
|
|
async def call(self, ctx: Any, parameters: dict[str, Any]) -> MCPToolResult:
|
|
gateway_calls.append({"db": self.db, "ctx": ctx, "parameters": parameters})
|
|
return MCPToolResult(
|
|
success=True,
|
|
output={"tool": ctx.tool_name, "ok": True},
|
|
execution_id="fake-gateway-ssh-exec",
|
|
)
|
|
|
|
monkeypatch.setattr("src.services.approval_execution.get_redis", lambda: fake_redis)
|
|
monkeypatch.setattr("src.services.approval_execution.get_db_context", fake_db_context)
|
|
monkeypatch.setattr("src.services.approval_execution.McpGateway", FakeGateway)
|
|
|
|
approval = ApprovalRequest(
|
|
action="docker restart sentry-worker",
|
|
description="測試 SSH approved execution gateway audit",
|
|
risk_level=RiskLevel.LOW,
|
|
requested_by="test",
|
|
required_signatures=0,
|
|
incident_id="INC-TEST-GATEWAY",
|
|
)
|
|
|
|
result = await ApprovalExecutionService()._execute_ssh_host_action(
|
|
approval=approval,
|
|
host="192.168.0.110",
|
|
)
|
|
|
|
assert result.success is True
|
|
assert db_context_projects == ["awoooi"]
|
|
assert len(gateway_calls) == 1
|
|
|
|
call = gateway_calls[0]
|
|
ctx = call["ctx"]
|
|
assert ctx.project_id == "awoooi"
|
|
assert ctx.agent_id == "approval_executor"
|
|
assert ctx.tool_name == "ssh_docker_restart"
|
|
assert ctx.required_scope == "write"
|
|
assert ctx.is_shadow is False
|
|
assert ctx.environment == {"env": "prod"}
|
|
assert ctx.run_id == approval.id
|
|
assert isinstance(ctx.run_id, UUID)
|
|
assert ctx.trace_id == "INC-TEST-GATEWAY"
|
|
|
|
parameters = call["parameters"]
|
|
assert parameters["host"] == "192.168.0.110"
|
|
assert parameters["container_name"] == "sentry-worker"
|
|
assert parameters["trust_score"] == 0.85
|
|
assert parameters["_mcp_audit"]["agent_role"] == "approval_executor"
|
|
assert parameters["_mcp_audit"]["flywheel_node"] == "execute"
|
|
assert parameters["_mcp_audit"]["incident_id"] == "INC-TEST-GATEWAY"
|
|
|
|
assert fake_redis.set_calls == [
|
|
(
|
|
f"mcp_approval:awoooi:approval_executor:ssh_docker_restart:{approval.id}",
|
|
"approved",
|
|
600,
|
|
)
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_ssh_diagnose_approval_uses_gateway_without_gate5_projection(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_redis = FakeRedis()
|
|
gateway_calls: list[dict[str, Any]] = []
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(project_id: str | None = None):
|
|
yield {"project_id": project_id}
|
|
|
|
class FakeGateway:
|
|
def __init__(self, db: object) -> None:
|
|
self.db = db
|
|
|
|
async def call(self, ctx: Any, parameters: dict[str, Any]) -> MCPToolResult:
|
|
gateway_calls.append({"db": self.db, "ctx": ctx, "parameters": parameters})
|
|
return MCPToolResult(
|
|
success=True,
|
|
output={"tool": ctx.tool_name, "ok": True},
|
|
execution_id="fake-gateway-read-exec",
|
|
)
|
|
|
|
monkeypatch.setattr("src.services.approval_execution.get_redis", lambda: fake_redis)
|
|
monkeypatch.setattr("src.services.approval_execution.get_db_context", fake_db_context)
|
|
monkeypatch.setattr("src.services.approval_execution.McpGateway", FakeGateway)
|
|
|
|
approval = ApprovalRequest(
|
|
action="df -h",
|
|
description="測試 SSH diagnose gateway audit",
|
|
risk_level=RiskLevel.LOW,
|
|
requested_by="test",
|
|
required_signatures=0,
|
|
incident_id="INC-TEST-READ",
|
|
)
|
|
|
|
result = await ApprovalExecutionService()._execute_ssh_host_action(
|
|
approval=approval,
|
|
host="192.168.0.110",
|
|
)
|
|
|
|
assert result.success is True
|
|
assert fake_redis.set_calls == []
|
|
assert len(gateway_calls) == 1
|
|
ctx = gateway_calls[0]["ctx"]
|
|
assert ctx.agent_id == "approval_executor"
|
|
assert ctx.tool_name == "ssh_diagnose"
|
|
assert ctx.required_scope == "read"
|
|
assert ctx.is_shadow is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gateway_block_returns_failed_execution_result_without_rollback(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
fake_redis = FakeRedis()
|
|
db_context_exits: list[str] = []
|
|
|
|
@asynccontextmanager
|
|
async def fake_db_context(project_id: str | None = None):
|
|
try:
|
|
yield {"project_id": project_id}
|
|
except Exception:
|
|
db_context_exits.append("raised")
|
|
raise
|
|
else:
|
|
db_context_exits.append("normal")
|
|
|
|
class FakeGateway:
|
|
def __init__(self, db: object) -> None:
|
|
self.db = db
|
|
|
|
async def call(self, ctx: Any, parameters: dict[str, Any]) -> MCPToolResult:
|
|
raise GateApprovalError("approval missing in smoke path")
|
|
|
|
monkeypatch.setattr("src.services.approval_execution.get_redis", lambda: fake_redis)
|
|
monkeypatch.setattr("src.services.approval_execution.get_db_context", fake_db_context)
|
|
monkeypatch.setattr("src.services.approval_execution.McpGateway", FakeGateway)
|
|
|
|
approval = ApprovalRequest(
|
|
action="docker restart sentry-worker",
|
|
description="測試 Gateway blocked audit 不被 rollback",
|
|
risk_level=RiskLevel.LOW,
|
|
requested_by="test",
|
|
required_signatures=0,
|
|
incident_id="INC-TEST-BLOCKED",
|
|
)
|
|
|
|
result = await ApprovalExecutionService()._execute_ssh_host_action(
|
|
approval=approval,
|
|
host="192.168.0.110",
|
|
)
|
|
|
|
assert result.success is False
|
|
assert result.error is not None
|
|
assert "E-MCP-GATE-005" in result.error
|
|
assert db_context_exits == ["normal"]
|
|
assert fake_redis.set_calls == [
|
|
(
|
|
f"mcp_approval:awoooi:approval_executor:ssh_docker_restart:{approval.id}",
|
|
"approved",
|
|
600,
|
|
)
|
|
]
|