feat(owner-gate): add allowlist-only maintenance runner

runtime/owner_maintenance/allowlist_runner.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Owner maintenance allowlist runner.
+
+Executes only approved command_key mappings. It never runs arbitrary shell strings.
+"""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import subprocess
+import sys
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+try:
+    import yaml  # type: ignore
+except Exception:
+    yaml = None
+
+
+@dataclass(frozen=True)
+class AllowedCommand:
+    command_key: str
+    argv: tuple[str, ...]
+    permission: str
+    risk: str
+    destructive: bool
+    timeout_seconds: int
+
+
+ALLOWED_COMMANDS: dict[str, AllowedCommand] = {
+    "openclaw.memory.index.force": AllowedCommand(
+        command_key="openclaw.memory.index.force",
+        argv=("/usr/local/bin/openclaw", "memory", "index", "--force"),
+        permission="OWNER_NON_DANGEROUS_ALL",
+        risk="medium",
+        destructive=False,
+        timeout_seconds=1800,
+    ),
+}
+
+DENY_FLAGS = {
+    "arbitrary_shell",
+    "credential_mutation",
+    "provider_mutation",
+    "external_network_write",
+    "production_data_delete",
+    "baseline_or_tag_mutation",
+    "main_branch_direct_write",
+}
+
+
+def sha256_text(value: str) -> str:
+    return hashlib.sha256(value.encode("utf-8", errors="replace")).hexdigest()
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def load_request(path: Path) -> dict[str, Any]:
+    if yaml is None:
+        raise RuntimeError("PyYAML is required to load owner request YAML")
+    data = yaml.safe_load(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError("owner request must be a YAML object")
+    return data
+
+
+def validate_request(req: dict[str, Any]) -> AllowedCommand:
+    command_key = req.get("command_key")
+    if command_key not in ALLOWED_COMMANDS:
+        raise PermissionError(f"command_key is not allowlisted: {command_key}")
+    allowed = ALLOWED_COMMANDS[command_key]
+    if req.get("permission") != allowed.permission:
+        raise PermissionError("request permission does not match allowlist")
+    safety = req.get("safety", {})
+    if not isinstance(safety, dict):
+        raise PermissionError("safety metadata is required")
+    if bool(safety.get("destructive", True)) != allowed.destructive:
+        raise PermissionError("destructive flag does not match allowlist")
+    denied = [flag for flag in DENY_FLAGS if safety.get(flag) is True]
+    if denied:
+        raise PermissionError(f"denied safety flags: {denied}")
+    gate = req.get("owner_gate", {})
+    if not isinstance(gate, dict) or gate.get("required") is not True:
+        raise PermissionError("owner_gate.required must be true")
+    if not gate.get("approved_by") or not gate.get("approved_at") or not gate.get("approval_commit"):
+        raise PermissionError("owner approval fields are required before execution")
+    return allowed
+
+
+def run_allowed_command(allowed: AllowedCommand, dry_run: bool) -> dict[str, Any]:
+    if dry_run:
+        return {"dry_run": True, "argv": list(allowed.argv), "exit_code": None}
+    proc = subprocess.run(
+        list(allowed.argv),
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=allowed.timeout_seconds,
+    )
+    return {
+        "dry_run": False,
+        "argv": list(allowed.argv),
+        "exit_code": proc.returncode,
+        "executed_at": utc_now(),
+        "stdout_sha256": sha256_text(proc.stdout),
+        "stderr_sha256": sha256_text(proc.stderr),
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Run an owner-approved allowlisted maintenance request")
+    parser.add_argument("request", type=Path)
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+    try:
+        req = load_request(args.request)
+        allowed = validate_request(req)
+        result = run_allowed_command(allowed, dry_run=args.dry_run)
+        print(json.dumps({"status": "ok", "command_key": allowed.command_key, "result": result}, indent=2))
+        return 0 if result.get("exit_code") in (None, 0) else int(result["exit_code"])
+    except Exception as exc:
+        print(json.dumps({"status": "error", "error": str(exc)}, indent=2), file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())