3156ff1c69
Adds Group B SSH MCP tool ssh_docker_prune (image+volume+builder prune with ≥75% disk usage gate) and routes "docker prune" actions through it. Flips HostDiskUsageHigh from auto_repair=false to true with mcp_provider routing labels so the flywheel can self-heal next disk-full event without hitting the emergency_channel Telegram path. Trigger: 2026-05-01 → 05-02 Telegram alert storm (peak 53/hr) caused by empty ssh-mcp-key/known_hosts secret rejecting all SSH and forcing every disk-full alert through "Host key is not trusted → escalate" loop. known_hosts patched live; this commit closes the playbook gap so the next occurrence resolves without manual intervention. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""
|
||
ssh_docker_prune 工具測試
|
||
==========================
|
||
ADR-068 飛輪 — disk full SOP（2026-05-02 ogt + Claude Sonnet 4.6）
|
||
|
||
測試項目：
|
||
- ssh_docker_prune 在 GROUP_B_TOOLS 白名單內
|
||
- list_tools() 回傳含 ssh_docker_prune 且 schema 正確
|
||
- _build_command() 產生包含 75% 磁碟守衛 + image/volume/builder prune 鏈
|
||
- _build_command() 不會引入 FORBIDDEN_PATTERNS（rm -rf, $( ), backtick 等）
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
|
||
from src.plugins.mcp.providers.ssh_provider import (
|
||
DOCKER_PRUNE_DISK_GATE_PCT,
|
||
GROUP_B_TOOLS,
|
||
SSHProvider,
|
||
)
|
||
|
||
|
||
class TestDockerPruneToolRegistration:
|
||
"""白名單與 list_tools() 註冊"""
|
||
|
||
def test_in_group_b(self):
|
||
assert "ssh_docker_prune" in GROUP_B_TOOLS
|
||
|
||
def test_disk_gate_constant(self):
|
||
# 守衛閾值合理（不會太低誤觸，不會太高永遠不跑）
|
||
assert 50 <= DOCKER_PRUNE_DISK_GATE_PCT <= 90
|
||
|
||
def test_listed_with_schema(self):
|
||
provider = SSHProvider()
|
||
tools = asyncio.run(provider.list_tools())
|
||
prune = next((t for t in tools if t.name == "ssh_docker_prune"), None)
|
||
assert prune is not None, "ssh_docker_prune not registered in list_tools()"
|
||
assert prune.server_name == provider.name
|
||
assert "host" in prune.input_schema["required"]
|
||
assert "trust_score" in prune.input_schema["required"]
|
||
|
||
|
||
class TestDockerPruneCommand:
|
||
"""命令字串建構"""
|
||
|
||
def _cmd(self) -> str:
|
||
provider = SSHProvider()
|
||
return provider._build_command("ssh_docker_prune", {"host": "192.168.0.110"})
|
||
|
||
def test_includes_disk_gate(self):
|
||
cmd = self._cmd()
|
||
# 應該先檢查磁碟使用率，低於閾值就 skip
|
||
assert "df --output=pcent" in cmd
|
||
assert f"{DOCKER_PRUNE_DISK_GATE_PCT}" in cmd
|
||
assert "skip" in cmd
|
||
|
||
def test_includes_full_prune_chain(self):
|
||
cmd = self._cmd()
|
||
assert "docker image prune -a -f" in cmd
|
||
assert "docker volume prune -f" in cmd
|
||
assert "docker builder prune -a -f" in cmd
|
||
|
||
def test_reports_disk_before_and_after(self):
|
||
cmd = self._cmd()
|
||
assert "DISK BEFORE" in cmd
|
||
assert "DISK AFTER" in cmd
|
||
|
||
def test_no_destructive_outside_docker(self):
|
||
"""命令只該動 docker 子系統，不該有 rm -rf / 寫 /etc / 改 sudoers"""
|
||
cmd = self._cmd()
|
||
assert "rm -rf" not in cmd
|
||
assert "/etc/" not in cmd
|
||
assert "sudoers" not in cmd
|
||
assert "authorized_keys" not in cmd
|
||
|
||
def test_only_uses_safe_params(self):
|
||
"""ssh_docker_prune 不接受 user-supplied 字串 params（只有 host/trust_score）"""
|
||
provider = SSHProvider()
|
||
tools = asyncio.run(provider.list_tools())
|
||
prune = next(t for t in tools if t.name == "ssh_docker_prune")
|
||
# schema 只有 host (string, whitelist-checked) + trust_score (number)
|
||
assert set(prune.input_schema["properties"].keys()) == {"host", "trust_score"}
|