fix(solver): 補 import re(solver_agent 已有 re.compile 但漏 import)
Some checks are pending
CD Pipeline / build-and-deploy (push) Has started running

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-04-25 02:42:25 +08:00
parent a49554c5a0
commit 39f45dd305
2 changed files with 291 additions and 16 deletions

View File

@@ -21,6 +21,7 @@ from __future__ import annotations
import asyncio
import hashlib
import re
import time
from typing import Any
@@ -41,6 +42,32 @@ logger = structlog.get_logger(__name__)
# Phase 2 單步 LLM timeout保留 Critic/Coordinator 的全局預算)
PHASE2_STEP_TIMEOUT_SEC = 20.0
# 2026-04-24 ogt + Claude Sonnet 4.6: kubectl 白名單正則Major #1 改版)
# 根因:黑名單枚舉不完整(如 $VAR、%0a、反引號 unicode 等繞過向量)
# 修復:改為白名單正則,只允許 kubectl 合法字元集
# 合法字符:英數、空白、- = . / : _ , @ (足以覆蓋完整 kubectl 語法)
# 任何不在此集合的字符(; & | ` $ > < 換行等)直接拒絕
# 範圍Nemo 路徑 + 標準 candidates 路徑雙層防護
_KUBECTL_COMMAND_PATTERN = re.compile(r"^kubectl\s+[A-Za-z0-9\s\-=./:_,@]+$")
def _is_safe_kubectl_command(cmd: str) -> bool:
"""kubectl 命令白名單驗證。
只允許 kubectl 開頭 + 合法字符集(英數、空白、- = . / : _ , @)。
任何 shell 元字符(; & | ` $ > < 換行等)皆返回 False。
Args:
cmd: 待驗證的命令字串
Returns:
True — 通過白名單False — 含非法字符或非 kubectl 開頭
"""
cmd = str(cmd).strip()
if not cmd.startswith("kubectl"):
return False
return _KUBECTL_COMMAND_PATTERN.fullmatch(cmd) is not None
class SolverAgent(BaseAgent):
"""
@@ -312,7 +339,19 @@ def _extract_candidates(parsed: dict[str, Any]) -> list[CandidateAction]:
# 新:先嘗試語意合成 kubectl 指令;真的無從映射才 return []
if "action_title" in parsed and "candidates" not in parsed:
action_title = str(parsed.get("action_title", ""))
confidence = float(parsed.get("confidence", 0.5))
# 2026-04-24 ogt + Claude Sonnet 4.6: confidence try/except + clampMajor #3/#4
# 根因LLM 可能回傳非數字字串(如 "high")或超界值(如 1.5 / -0.1
# float() 直接呼叫會 ValueError超界值會破壞 auto_approve 門檻判斷
# 修復try/except 捕捉型別錯誤 → 預設 0.5;再 clamp 到 [0.0, 1.0]
try:
confidence = float(parsed.get("confidence", 0.5))
except (TypeError, ValueError):
logger.warning(
"solver_nemo_confidence_type_error",
raw_confidence=parsed.get("confidence"),
)
confidence = 0.5
confidence = max(0.0, min(1.0, confidence))
risk_level = str(parsed.get("risk_level", "medium"))
risk_to_blast = {"critical": 60, "high": 40, "medium": 25, "low": 10}
blast = risk_to_blast.get(risk_level.lower(), 30)
@@ -323,19 +362,30 @@ def _extract_candidates(parsed: dict[str, Any]) -> list[CandidateAction]:
# 修法 A優先採用 kubectl_command 欄位,保留原始 confidence如 0.9
kubectl_cmd = str(parsed.get("kubectl_command", "")).strip()
if kubectl_cmd and kubectl_cmd.startswith("kubectl"):
logger.debug(
"solver_nemo_kubectl_command_used",
action_title=action_title[:80],
kubectl_command=kubectl_cmd[:80],
confidence=confidence,
)
return [CandidateAction(
action=kubectl_cmd[:200],
blast_radius=blast,
rollback_cost=20,
confidence=confidence,
rationale=f"OpenClaw Nemo: {action_title[:80]}",
)]
# 2026-04-24 ogt + Claude Sonnet 4.6: 白名單正則防注入Major #1 改版)
# 根因黑名單枚舉不完整改用白名單正則_is_safe_kubectl_command
# 修復:不通過白名單 → log warning → fall-through 到 action_title 路徑(不 return
if not _is_safe_kubectl_command(kubectl_cmd):
logger.warning(
"solver_kubectl_invalid_syntax",
cmd=kubectl_cmd[:80],
reason="未通過白名單正則檢驗",
)
# fall-through不採用此 kubectl_command繼續往下走 action_title 路徑
else:
logger.debug(
"solver_nemo_kubectl_command_used",
action_title=action_title[:80],
kubectl_command=kubectl_cmd[:80],
confidence=confidence,
)
return [CandidateAction(
action=kubectl_cmd[:200],
blast_radius=blast,
rollback_cost=20,
confidence=confidence,
rationale=f"OpenClaw Nemo: {action_title[:80]}",
)]
if "kubectl" in action_title.lower():
if action_title and confidence > 0:
@@ -389,8 +439,19 @@ def _extract_candidates(parsed: dict[str, Any]) -> list[CandidateAction]:
for item in raw:
if not isinstance(item, dict):
continue
# 2026-04-24 ogt + Claude Sonnet 4.6: 標準 candidates 路徑白名單防護Major #2
# 根因:標準路徑未驗證 action 欄位LLM 可注入含 shell 元字符的惡意命令
# 修復:每個 action 通過 _is_safe_kubectl_command 白名單檢驗,失敗則跳過
action = str(item.get("action", ""))[:200]
if not _is_safe_kubectl_command(action):
logger.warning(
"solver_standard_action_unsafe",
action=action[:80],
reason="未通過白名單檢驗",
)
continue
c = CandidateAction(
action=str(item.get("action", ""))[:200],
action=action,
blast_radius=max(0, min(100, int(item.get("blast_radius", 50)))),
rollback_cost=max(0, min(100, int(item.get("rollback_cost", 50)))),
confidence=float(item.get("confidence", 0.0)),