✏️ 正在编辑: openclaw.py
路径:
/var/opt/nydus/ops/customer_local_ops/openclaw/openclaw.py
提示:
您可以编辑任何文件（包括二进制文件），但请注意不当修改可能导致文件损坏。
"""Nydus ops for installing and managing OpenClaw on customer VMs.""" import json import logging import os import re from typing import Any, Dict, List, Optional, Tuple from urllib.parse import urlparse from customer_local_ops import Ops, NydusResult from customer_local_ops.util.execute import runCommand LOG = logging.getLogger(__name__) DEFAULT_INSTALLER_BASE = "https://hfs-public.secureserver.net" INSTALLER_PATH_SUFFIX = "/-/ubuntu/amd64/24.04/executable/openclaw-installer-latest.signed" IMAGE_PATH_SUFFIX = "/-/ubuntu/amd64/24.04/container/docker/openclaw-latest.tar.gz" INSTALLER_PATH = "/tmp/openclaw-installer" SIGNED_DOWNLOAD_PATH = "/tmp/openclaw-installer.signed" SIG_TEMP_PATH = "/tmp/openclaw-installer.sig.asc" SIGNED_FILE_MARKER = b"\n---OPENCLAW-SIG---\n" ENV_FILE_PATH = "/opt/openclaw/.env" CONFIG_JSON_PATH = "/opt/openclaw/config/openclaw.json" PUB_KEY_PATH_SUFFIX = "/-/gpg/nydus-signing.pub.asc" PUB_KEY_DOWNLOAD_PATH = "/tmp/nydus-signing.pub.asc" _CONTAINER_NAME_RE = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_.-]*$") _INSTALLER_ARG_RE = re.compile(r"^[^\s;&|`$<>\"']*$") _REDACTED = "<<REMOVED>>" def _minimal_log_entry(entry): """Return only step name and success for completion callback (keeps payload small).""" return {"step": entry.get("step", ""), "success": entry.get("success", False)} def _redact(text, secrets): """Replace each non-empty secret in *text* with a redaction marker.""" for s in secrets: if s: text = text.replace(s, _REDACTED) return text def _run_step(cmd, tag, step_name, omit_strings=None): """Run a shell command and return a structured log entry.""" primary_omit = omit_strings[0] if omit_strings else None exit_code, outs, errs = runCommand(cmd, tag, useShell=True, omitString=primary_omit) log_entry = { "step": step_name, "success": exit_code == 0, "stdout": outs, "stderr": errs, } if omit_strings: log_entry["stdout"] = _redact(log_entry["stdout"], omit_strings) log_entry["stderr"] = _redact(log_entry["stderr"], omit_strings) return exit_code, outs, errs, log_entry def _run_step_argv(cmd_list, tag, step_name, omit_strings=None): """Run a command as an argv list (no shell) and return a structured log entry.""" primary_omit = omit_strings[0] if omit_strings else None exit_code, outs, errs = runCommand(cmd_list, tag, useShell=False, omitString=primary_omit) log_entry = { "step": step_name, "success": exit_code == 0, "stdout": outs, "stderr": errs, } if omit_strings: log_entry["stdout"] = _redact(log_entry["stdout"], omit_strings) log_entry["stderr"] = _redact(log_entry["stderr"], omit_strings) return exit_code, outs, errs, log_entry def _validate_installer_arg(value, field_name): """Validate an installer argument; reject shell metacharacters.""" if value is None: return "" value_str = str(value) if not _INSTALLER_ARG_RE.match(value_str): raise ValueError(f"Invalid characters in {field_name}") return value_str def _validate_base_url(raw_url): """Validate and normalize an installer base URL (must have https/http scheme and a host).""" url = raw_url.strip().rstrip("/") parsed = urlparse(url) if parsed.scheme not in ("https", "http"): raise ValueError( f"installer_base_url must use https or http scheme, got: {parsed.scheme!r}") if not parsed.netloc: raise ValueError("installer_base_url must include a hostname") return f"{parsed.scheme}://{parsed.netloc}{parsed.path}" def _install_fail(logs: List[dict], step_failed: str) -> NydusResult: return False, json.dumps({ "logs": logs, "step_failed": step_failed, "setup_complete": False, }) def _resolve_urls(payload_dict: Dict[str, Any]) -> Tuple[str, str, str]: """Return (installer_url, pub_key_url, image_url) from the validated base URL.""" raw_base = ( payload_dict.get("installer_base_url") or payload_dict.get("pypi_url") or DEFAULT_INSTALLER_BASE ) base = _validate_base_url(raw_base) installer_url = base + INSTALLER_PATH_SUFFIX pub_key_url = base + PUB_KEY_PATH_SUFFIX image_url = base + IMAGE_PATH_SUFFIX return installer_url, pub_key_url, image_url def _download_signed_installer(logs: List[dict], installer_url: str) -> Optional[NydusResult]: exit_code, _, _, log_entry = _run_step_argv( ["curl", "-fSL", "-o", SIGNED_DOWNLOAD_PATH, installer_url], "openclaw_download", "download_signed_installer") logs.append(log_entry) if exit_code != 0: return _install_fail(logs, "download_signed_installer") return None def _download_public_key(logs: List[dict], pub_key_url: str) -> Optional[NydusResult]: exit_code, _, _, log_entry = _run_step_argv( ["curl", "-fSL", "-o", PUB_KEY_DOWNLOAD_PATH, pub_key_url], "openclaw_download_pubkey", "download_pubkey") logs.append(log_entry) if exit_code != 0: return _install_fail(logs, "download_pubkey") return None def _extract_installer_from_signed_file(logs: List[dict]) -> Optional[NydusResult]: try: with open(SIGNED_DOWNLOAD_PATH, "rb") as handle: data_bytes = handle.read() idx = data_bytes.find(SIGNED_FILE_MARKER) if idx < 0: logs.append({ "step": "extract_signed", "success": False, "stderr": "Marker not found in .signed file", }) return _install_fail(logs, "extract_signed") with open(INSTALLER_PATH, "wb") as handle: handle.write(data_bytes[:idx]) with open(SIG_TEMP_PATH, "wb") as handle: handle.write(data_bytes[idx + len(SIGNED_FILE_MARKER):]) except OSError as exc: logs.append({ "step": "extract_signed", "success": False, "stderr": str(exc), }) return _install_fail(logs, "extract_signed") logs.append({ "step": "extract_signed", "success": True, "stdout": "", "stderr": "", }) return None def _verify_signed_installer_with_gpg(logs: List[dict]) -> Optional[NydusResult]: steps = ( ( "command -v gpg >/dev/null 2>&1 || " "(apt-get update -qq && apt-get install -y -qq gnupg)", "openclaw_ensure_gpg", "ensure_gpg", "ensure_gpg", ), ( f"gpg --batch --import {PUB_KEY_DOWNLOAD_PATH}", "openclaw_gpg_import", "gpg_import", "gpg_import", ), ( f"gpg --batch --verify {SIG_TEMP_PATH} {INSTALLER_PATH}", "openclaw_gpg_verify", "gpg_verify", "gpg_verify", ), ( f"chmod +x {INSTALLER_PATH}", "openclaw_chmod", "chmod_installer", "chmod_installer", ), ) for cmd, tag, step_name, step_key in steps: exit_code, _, _, log_entry = _run_step(cmd, tag, step_name) logs.append(log_entry) if exit_code != 0: return _install_fail(logs, step_key) return None def _cleanup_installer_artifacts(): """Remove signed bundle, signature, and public key left over from installer verification.""" for path in (SIGNED_DOWNLOAD_PATH, SIG_TEMP_PATH, PUB_KEY_DOWNLOAD_PATH): try: os.remove(path) except OSError: pass def _run_installer_binary_and_check_env(logs: List[dict], api_key: str, telegram_token: str, email: str, image_url: str = "") -> NydusResult: sensitive = [v for v in (api_key, telegram_token) if v] installer_cmd = [ "sudo", INSTALLER_PATH, "install", f"--api-key={api_key}", "--https", f"--telegram-token={telegram_token}", f"--email={email}", ] if image_url: installer_cmd.append(f"--image-url={image_url}") exit_code, _, _, log_entry = _run_step_argv( installer_cmd, "openclaw_install", "run_installer", omit_strings=sensitive) logs.append(log_entry) install_ok = exit_code == 0 _, v_outs, _, verify_entry = _run_step( f"test -f {ENV_FILE_PATH} && echo 'VERIFIED' || echo 'NOT_FOUND'", "openclaw_verify", "verify_installation", ) verify_entry["success"] = "VERIFIED" in v_outs logs.append(verify_entry) setup_complete = install_ok and "VERIFIED" in v_outs data = { "logs": [_minimal_log_entry(e) for e in logs], "setup_complete": setup_complete, "status": "complete" if setup_complete else "failed", } if not setup_complete: data["step_failed"] = ( "run_installer" if not install_ok else "verify_installation") return setup_complete, json.dumps(data) class OpenClaw(Ops): # pylint: disable=abstract-method """Nydus ops for installing and managing OpenClaw on customer VMs.""" def install_openclaw(self, payload) -> NydusResult: """Download signed installer, verify with GPG, install, and verify OpenClaw.""" logs: List[dict] = [] pl = payload or {} try: api_key = _validate_installer_arg(pl.get("api_key", ""), "api_key") telegram_token = _validate_installer_arg( pl.get("telegram_token", ""), "telegram_token") email = _validate_installer_arg(pl.get("email", ""), "email") installer_url, pub_key_url, image_url = _resolve_urls(pl) except ValueError as exc: data = {"logs": [], "step_failed": "validation", "setup_complete": False, "error": str(exc)} return False, json.dumps(data) try: for step_fn in ( lambda: _download_signed_installer(logs, installer_url), lambda: _download_public_key(logs, pub_key_url), lambda: _extract_installer_from_signed_file(logs), lambda: _verify_signed_installer_with_gpg(logs), ): err = step_fn() if err is not None: return err finally: _cleanup_installer_artifacts() return _run_installer_binary_and_check_env( logs, api_key, telegram_token, email, image_url=image_url) def update_openclaw(self, payload) -> NydusResult: """Re-download the installer binary, verify with GPG, then run ``openclaw-installer update`` to replace the Docker image.""" logs: List[dict] = [] pl = payload or {} try: installer_url, pub_key_url, image_url = _resolve_urls(pl) except ValueError as exc: data = {"logs": [], "step_failed": "validation", "setup_complete": False, "error": str(exc)} return False, json.dumps(data) try: for step_fn in ( lambda: _download_signed_installer(logs, installer_url), lambda: _download_public_key(logs, pub_key_url), lambda: _extract_installer_from_signed_file(logs), lambda: _verify_signed_installer_with_gpg(logs), ): err = step_fn() if err is not None: return err finally: _cleanup_installer_artifacts() cmd = ["sudo", INSTALLER_PATH, "update"] if image_url: cmd.append(f"--image-url={image_url}") exit_code, _, _, log_entry = _run_step_argv( cmd, "openclaw_update", "run_update") logs.append(log_entry) data = { "logs": [_minimal_log_entry(e) for e in logs], "setup_complete": exit_code == 0, "status": "complete" if exit_code == 0 else "failed", } if exit_code != 0: data["step_failed"] = "run_update" return exit_code == 0, json.dumps(data) def get_openclaw_setup_status(self) -> NydusResult: """Check if OpenClaw env file and docker containers are present.""" logs = [] env_cmd = f"test -f {ENV_FILE_PATH} && echo 'exists' || echo 'missing'" _, outs, _, log_entry = _run_step( env_cmd, "openclaw_env_check", "env_file_check") env_exists = "exists" in outs log_entry["success"] = env_exists logs.append(log_entry) docker_cmd = ( "sudo docker ps --filter 'name=openclaw'" " --format '{{.Names}}:{{.Status}}'") exit_code, outs, errs, log_entry = _run_step( docker_cmd, "openclaw_docker_check", "docker_containers_check") containers_running = exit_code == 0 and len(outs.strip()) > 0 log_entry["success"] = containers_running logs.append(log_entry) if exit_code != 0: data = { "setup_complete": False, "env_file_exists": env_exists, "containers": outs.strip(), "error": "docker ps failed: " + errs.strip(), "logs": [_minimal_log_entry(e) for e in logs], } return False, json.dumps(data) data = { "setup_complete": env_exists and containers_running, "env_file_exists": env_exists, "containers": outs.strip(), "logs": [_minimal_log_entry(e) for e in logs], } return True, json.dumps(data) def get_openclaw_gateway_token(self) -> NydusResult: """Extract gateway token from openclaw.json (single source of truth).""" try: with open(CONFIG_JSON_PATH, "r", encoding="utf-8") as fh: cfg = json.load(fh) token_value = (cfg.get("gateway", {}) .get("auth", {}) .get("token", "")) except (OSError, json.JSONDecodeError) as exc: data = {"token": None, "error": str(exc)} return False, json.dumps(data) if not token_value: data = {"token": None, "error": "gateway.auth.token missing in openclaw.json"} return False, json.dumps(data) data = {"token": token_value} return True, json.dumps(data) def manage_openclaw_containers(self, payload) -> NydusResult: """List, start, or stop OpenClaw docker containers.""" action = payload.get("action", "list") container_name = payload.get("container_name", "") if action == "list": cmd = ("sudo docker ps -a --filter 'name=openclaw'" " --format '{{json .}}'") exit_code, outs, errs = runCommand( cmd, "openclaw_docker_list", useShell=True) containers = [] for line in outs.strip().splitlines(): if line.strip(): try: containers.append(json.loads(line)) except json.JSONDecodeError: containers.append({"raw": line}) data = {"action": action, "containers": containers, "error": errs.strip()} return exit_code == 0, json.dumps(data) if action in ("start", "stop"): if not container_name: data = {"action": action, "error": "container_name is required for start/stop"} return False, json.dumps(data) if not _CONTAINER_NAME_RE.match(container_name): data = {"action": action, "container_name": container_name, "error": "Invalid container_name; only letters, " "digits, '.', '-', and '_' are allowed"} return False, json.dumps(data) cmd = ["sudo", "docker", action, container_name] exit_code, outs, errs = runCommand( cmd, f"openclaw_docker_{action}", useShell=False) data = {"action": action, "container_name": container_name, "result": outs.strip(), "error": errs.strip()} return exit_code == 0, json.dumps(data) data = {"action": action, "error": f"Unknown action: {action}"} return False, json.dumps(data)
💾 保存文件
← 返回文件管理器