feat(api): URI scheme 解析器 + Shell Injection 防護 (Sprint 3 T1)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-04-06 14:18:21 +08:00
parent 9197994d51
commit 5e8b2a6894
2 changed files with 135 additions and 0 deletions

View File

@@ -35,6 +35,61 @@ _COMPONENT_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,30}$")
SSH_TIMEOUT = 60 # seconds
# =============================================================================
# URI Scheme 解析
# 2026-04-06 Claude Code: Sprint 3 T1
# =============================================================================
@dataclass
class SshCommandURI:
"""解析後的 SSH_COMMAND URI"""
scheme: str # "openclaw" | "ansible" | "ssh"
host_or_layer: str # "docker-110" | "192.168.0.188" | "wooo@192.168.0.110"
payload: str # component name | playbook filename | raw command
_SUPPORTED_SCHEMES = {"openclaw", "ansible", "ssh"}
_SHELL_METACHAR_RE = re.compile(r'[;&|`&]|\$\(')
_MAX_COMMAND_LEN = 512
def parse_uri_command(command: str) -> SshCommandURI:
"""
解析 SSH_COMMAND URI scheme。
支援格式:
openclaw://docker-110/sentry
ansible://192.168.0.188/vacuum_postgres.yml
ssh://wooo@192.168.0.110/docker ps
Raises:
ValueError: scheme 不支援或 payload 為空
"""
if "://" not in command:
raise ValueError(f"Unsupported scheme: '{command}' (expected scheme://host/payload)")
scheme, rest = command.split("://", 1)
if scheme not in _SUPPORTED_SCHEMES:
raise ValueError(f"Unsupported scheme: '{scheme}' (supported: {_SUPPORTED_SCHEMES})")
if "/" not in rest:
raise ValueError(f"Invalid URI '{command}': missing payload after host")
host_or_layer, payload = rest.split("/", 1)
if not payload:
raise ValueError(f"Invalid URI '{command}': payload is empty")
return SshCommandURI(scheme=scheme, host_or_layer=host_or_layer, payload=payload)
def validate_shell_safety(command: str) -> None:
"""
驗證 ssh:// payload 不含 shell metacharacter 或超長命令。
Raises:
ValueError: 含危險字元或超過長度限制
"""
if len(command) > _MAX_COMMAND_LEN:
raise ValueError(f"Command too long: {len(command)} > {_MAX_COMMAND_LEN}")
if _SHELL_METACHAR_RE.search(command):
raise ValueError(f"Shell metacharacter detected in command: '{command}'")
@dataclass
class HostRepairResult: