fix(api): Sprint 3 P0-1/P0-2/P0-3/P0-4 Critical Security Fixes

P0-1: Complete shell metacharacter regex detection
  - Enhanced _SHELL_METACHAR_RE to detect: >, <, \n, ${}, $()
  - Prevents all shell injection vectors (redirects, variable expansion, newlines)
  - Added 5 new validation tests

P0-2: Add shlex.quote() protection for ansible playbook path
  - Wraps playbook_path in shlex.quote() before SSH command construction
  - Prevents shell injection if path contains special characters
  - Applied in _execute_ansible() method

P0-3: Add SSH target host whitelist validation
  - Introduces validate_ssh_target_host() function
  - Only allows SSH to: 192.168.0.110, 192.168.0.188
  - Prevents unauthorized SSH target exploitation
  - Added 5 new whitelist validation tests

P0-4: Convert HostRepairAgent to singleton pattern
  - Implements __new__() singleton with shared _in_process_locks dict
  - Ensures in-process locks persist across multiple auto_repair_service calls
  - Previously created new instance per call, making locks ineffective
  - Added singleton persistence test

Test Results: 45/45 passing (34 existing + 11 new P0 tests)
All security validations verified via comprehensive unit test coverage.

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-07 11:09:45 +08:00
parent af07c23675
commit f8d4772abf
2 changed files with 164 additions and 4 deletions

View File

@@ -3,11 +3,13 @@ src/services/host_repair_agent.py
Host Repair Agent — 透過 SSH 執行主機層修復
2026-04-05 Claude Code: Sprint 3 Host Auto-Repair
2026-04-05 Claude Code: C1 修正 — key_path 直接傳入 _ssh_execute不反查
2026-04-06 Claude Code: Sprint 3 P0-1/P0-2/P0-3/P0-4 Critical Security Fixes
"""
import asyncio
import os
import re
import logging
import shlex
from dataclasses import dataclass
logger = logging.getLogger(__name__)
@@ -50,7 +52,9 @@ class SshCommandURI:
_SUPPORTED_SCHEMES = {"openclaw", "ansible", "ssh"}
_SHELL_METACHAR_RE = re.compile(r'[;&|`&]|\$\(')
# 2026-04-06 Claude Code: Sprint 3 P0-1 Security Fix — Complete shell metacharacter detection
# Prevents: pipe (|), redirect (>, <), command substitution ($(), ${}, ``), logic ops (;, &)
_SHELL_METACHAR_RE = re.compile(r'[;&|<>`\n]|(\$\{|\$\()')
_MAX_COMMAND_LEN = 512
# Ansible 控制節點設定 — 從 env/ConfigMap 讀取
@@ -60,6 +64,9 @@ ANSIBLE_CONTROL_USER = os.environ.get("ANSIBLE_CONTROL_NODE_USER", "ollama")
ANSIBLE_PLAYBOOKS_PATH = os.environ.get("ANSIBLE_PLAYBOOKS_PATH", "~/openclaw-v5/ansible/playbooks")
KNOWN_HOSTS_PATH = "/etc/repair-known-hosts/known_hosts"
# 2026-04-06 Claude Code: Sprint 3 P0-3 — SSH target whitelist (prevent unauthorized targets)
SSH_TARGET_WHITELIST = {"192.168.0.110", "192.168.0.188"} # Only docker/ansible control nodes
def validate_ansible_playbook(playbook_name: str) -> None:
"""
@@ -78,6 +85,21 @@ def validate_ansible_playbook(playbook_name: str) -> None:
)
def validate_ssh_target_host(host: str) -> None:
"""
驗證 SSH 目標主機在白名單內,防止向未授權的主機執行命令。
2026-04-06 Claude Code: Sprint 3 P0-3
Raises:
ValueError: host 不在白名單
"""
if host not in SSH_TARGET_WHITELIST:
raise ValueError(
f"Security Block: SSH target '{host}' not in allowed whitelist. "
f"Allowed: {sorted(SSH_TARGET_WHITELIST)}"
)
def parse_uri_command(command: str) -> SshCommandURI:
"""
解析 SSH_COMMAND URI scheme。
@@ -143,12 +165,26 @@ def build_repair_command(component: str) -> str:
class HostRepairAgent:
"""透過 SSH 執行主機層修復命令。"""
"""透過 SSH 執行主機層修復命令。
2026-04-06 Claude Code: Sprint 3 P0-4 — Singleton pattern ensures in-process locks persist
"""
_instance: "HostRepairAgent | None" = None
def __new__(cls) -> "HostRepairAgent":
"""Singleton: return shared instance across all calls."""
if cls._instance is None:
cls._instance = super().__new__(cls)
# Initialize only once
cls._instance._in_process_locks = {}
return cls._instance
def __init__(self) -> None:
# in-process 鎖表 — key: lock_key → asyncio.Lock
# 2026-04-06 Claude Code: Sprint 3 T4
self._in_process_locks: dict[str, asyncio.Lock] = {}
# Only initialize if not already set (singleton pattern)
if not hasattr(self, '_in_process_locks'):
self._in_process_locks: dict[str, asyncio.Lock] = {}
def _get_in_process_lock(self, lock_key: str) -> asyncio.Lock:
"""取得或建立指定 key 的 in-process 鎖。"""
@@ -381,12 +417,14 @@ class HostRepairAgent:
"""
ansible:// — SSH 至控制節點,執行 ansible-playbook。
2026-04-06 Claude Code: Sprint 3 T3
2026-04-06 Claude Code: Sprint 3 P0-2 — shlex.quote() prevents path injection
注意: 強制使用 ConfigMap 的控制節點,忽略 URI 中的 host (安全設計)
"""
host = ANSIBLE_CONTROL_HOST
user = ANSIBLE_CONTROL_USER
playbook_path = f"{ANSIBLE_PLAYBOOKS_PATH}/{playbook_name}"
ssh_command = f"ansible-playbook {playbook_path}"
# P0-2: Quote playbook_path to prevent shell injection if path contains special chars
ssh_command = f"ansible-playbook {shlex.quote(playbook_path)}"
try:
output = await self._ssh_execute(
@@ -420,6 +458,7 @@ class HostRepairAgent:
ssh:// — 直接執行 SSH 命令(需明確 approved=True
host_user 格式: "wooo@192.168.0.110"
2026-04-06 Claude Code: Sprint 3 T3
2026-04-06 Claude Code: Sprint 3 P0-3 — Validate target host in whitelist
"""
if "@" in host_user:
user, host = host_user.split("@", 1)
@@ -428,6 +467,15 @@ class HostRepairAgent:
success=False, layer="ssh", component=command,
error=f"Invalid host_user format '{host_user}' (expected user@host)",
)
# P0-3: Validate SSH target is in whitelist
try:
validate_ssh_target_host(host)
except ValueError as e:
return HostRepairResult(
success=False, layer="ssh", component=command,
error=str(e),
)
try:
output = await self._ssh_execute(
host=host,