FM / openclaw

/var/opt/nydus/ops/customer_local_ops/openclaw/

/var/opt/nydus/ops/customer_local_ops/openclaw

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
openclaw.py	16559	0644	edit dl rm
__init__.py	163	0644	edit dl rm

Edit: /var/opt/nydus/ops/customer_local_ops/openclaw/openclaw.py (16559B)

"""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)

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)

cmd: