claudetools/.claude/scripts/backup_common.py

#!/usr/bin/env python3
"""Shared helpers for the backup-verification skills (seafile / owncloud / datto-workplace).

Three sibling skills answer the same GPS-audit question from different backends:
"which customers/machines are actually backing up here?" Each skill has its own
server-side client (Seafile Web API, ownCloud occ over SSH, Datto Workplace file
API). What they SHARE — and what lives here — is:

  * ClaudeTools repo-root resolution (env -> identity.json -> derived).
  * A one-field SOPS vault read via the canonical vault.sh wrapper.
  * An RmmClient for the "endpoint" half of the "server + RMM endpoint" cross-check:
    log in to GuruRMM, list agents, and run a detection PowerShell on an agent to
    prove a given sync client (SeaDrive / ownCloud / Datto Workplace) is actually
    installed and signed in on the machine.

Standalone: stdlib-only transport (urllib), no third-party dependency. Skills add
this file's directory to sys.path and `import backup_common`.
"""
from __future__ import annotations

import json
import os
import subprocess
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
from typing import Any, Optional

# backup_common.py lives at <root>/.claude/scripts/backup_common.py
_THIS = Path(__file__).resolve()
_DERIVED_ROOT = _THIS.parent.parent.parent  # scripts -> .claude -> repo root

RMM_BASE = os.environ.get("GURURMM_API", "http://172.16.3.30:3001")
RMM_VAULT_ENTRY = "infrastructure/gururmm-server.sops.yaml"
RMM_FIELD_EMAIL = "credentials.gururmm-api.admin-email"
RMM_FIELD_PASSWORD = "credentials.gururmm-api.admin-password"

HTTP_TIMEOUT = 60.0


class BackupError(RuntimeError):
    """Transport / auth / API error, with an optional HTTP status."""

    def __init__(self, message: str, *, status: Optional[int] = None):
        super().__init__(message)
        self.status = status


# --- repo-root + vault --------------------------------------------------------
def resolve_root() -> Path:
    """Resolve the ClaudeTools repo root: env var, then identity.json, then derived."""
    env_root = os.environ.get("CLAUDETOOLS_ROOT")
    if env_root:
        return Path(env_root)
    identity = _DERIVED_ROOT / ".claude" / "identity.json"
    if identity.exists():
        try:
            data = json.loads(identity.read_text(encoding="utf-8"))
            root = data.get("claudetools_root")
            if root:
                return Path(root)
        except (json.JSONDecodeError, OSError):
            pass
    return _DERIVED_ROOT


def vault_get_field(entry: str, field: str) -> str:
    """Read one field from a SOPS vault entry via the canonical vault.sh wrapper."""
    root = resolve_root()
    vault = root / ".claude" / "scripts" / "vault.sh"
    if not vault.exists():
        raise BackupError(f"vault wrapper not found at {vault}")
    try:
        done = subprocess.run(
            ["bash", str(vault), "get-field", entry, field],
            capture_output=True, text=True, timeout=60,
        )
    except FileNotFoundError as exc:
        raise BackupError("'bash' not found on PATH (need Git Bash for vault reads).") from exc
    except subprocess.TimeoutExpired as exc:
        raise BackupError(f"vault read timed out for {entry}:{field}") from exc
    if done.returncode != 0:
        raise BackupError(
            f"vault read failed for {entry}:{field} (exit {done.returncode}): {done.stderr.strip()}"
        )
    value = done.stdout.strip()
    if not value:
        raise BackupError(f"vault returned empty for {entry}:{field}")
    return value


# --- minimal JSON HTTP --------------------------------------------------------
def http_json(method: str, url: str, *, headers: Optional[dict] = None,
              body: Optional[dict] = None, form: Optional[dict] = None,
              timeout: float = HTTP_TIMEOUT,
              basic: Optional[tuple[str, str]] = None) -> tuple[int, Any]:
    """Do an HTTP request and parse a JSON response. Returns (status, parsed).

    `body` sends a JSON payload; `form` sends application/x-www-form-urlencoded
    (list values expand via doseq, e.g. repeated fields the Seafile admin share
    endpoint reads with getlist). Pass at most one. 4xx/5xx are returned (not
    raised) so callers can inspect the error body; transport failures raise.
    """
    hdrs = dict(headers or {})
    data = None
    if body is not None and form is not None:
        raise BackupError("http_json: pass body OR form, not both")
    if body is not None:
        data = json.dumps(body).encode("utf-8")
        hdrs.setdefault("Content-Type", "application/json")
    elif form is not None:
        data = urllib.parse.urlencode(form, doseq=True).encode("utf-8")
        hdrs.setdefault("Content-Type", "application/x-www-form-urlencoded")
    if basic is not None:
        import base64
        tok = base64.b64encode(f"{basic[0]}:{basic[1]}".encode()).decode("ascii")
        hdrs["Authorization"] = f"Basic {tok}"
    req = urllib.request.Request(url, data=data, method=method, headers=hdrs)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            raw = resp.read().decode("utf-8", errors="replace")
            return resp.getcode(), _parse(raw)
    except urllib.error.HTTPError as exc:
        raw = exc.read().decode("utf-8", errors="replace")
        return exc.code, _parse(raw)
    except urllib.error.URLError as exc:
        raise BackupError(f"request to {url} failed: {exc}") from exc


def _parse(text: str) -> Any:
    if not text:
        return {}
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        return {"_raw": text[:600]}


# --- GuruRMM client (endpoint cross-check) ------------------------------------
# One detection script, parametrized by the product name-fragments to look for.
# Returns a single JSON line describing installed products, running processes, and
# per-user config folders that match — enough to say "this client is installed and
# signed in" without guessing.
_DETECT_PS = r'''
$ErrorActionPreference='SilentlyContinue'
$patterns = @(__PATTERNS__)
function match($s){ if(-not $s){return $false}; foreach($p in $patterns){ if($s -match [regex]::Escape($p)){return $true} }; return $false }

$installed=@()
foreach($hive in @('HKLM:\Software\Microsoft\Windows\CurrentVersion\Uninstall\*','HKLM:\Software\WOW6432Node\Microsoft\Windows\CurrentVersion\Uninstall\*')){
  Get-ItemProperty $hive | Where-Object { match($_.DisplayName) } | ForEach-Object {
    $installed += [pscustomobject]@{ name=$_.DisplayName; version=$_.DisplayVersion; location=$_.InstallLocation }
  }
}
$procs = Get-Process | Where-Object { match($_.ProcessName) } | Select-Object -Expand ProcessName -Unique
$cfg=@()
foreach($d in @('Seafile','SeaDrive','ownCloud','ownCloudSync','Datto','Workplace','Workplace2')){
  foreach($base in @($env:APPDATA,$env:LOCALAPPDATA,$env:USERPROFILE,"$env:USERPROFILE\Desktop")){
    $pth = Join-Path $base $d
    if(Test-Path $pth){ if(match($d) -or match($pth)){ $cfg += $pth } }
  }
}
# signed-in heuristic: config/db files with recent writes
$fresh=@()
foreach($c in $cfg){
  Get-ChildItem $c -Recurse -File -ErrorAction SilentlyContinue -Depth 2 |
    Where-Object { $_.LastWriteTime -gt (Get-Date).AddDays(-14) } |
    Select-Object -First 1 | ForEach-Object { $fresh += $c }
}
[pscustomobject]@{
  host=$env:COMPUTERNAME
  installed=$installed
  processes=@($procs)
  config_dirs=@($cfg)
  active_config_dirs=@($fresh | Select-Object -Unique)
  detected=([bool]($installed.Count -or $procs.Count))
} | ConvertTo-Json -Depth 5 -Compress
'''


class RmmClient:
    """Thin GuruRMM API client: login, list agents, run a detection PowerShell."""

    def __init__(self, base: str = RMM_BASE):
        self.base = base.rstrip("/")
        self._token: Optional[str] = None
        self._agents: Optional[list[dict]] = None

    def login(self) -> None:
        email = vault_get_field(RMM_VAULT_ENTRY, RMM_FIELD_EMAIL)
        pw = vault_get_field(RMM_VAULT_ENTRY, RMM_FIELD_PASSWORD)
        status, body = http_json("POST", f"{self.base}/api/auth/login",
                                 body={"email": email, "password": pw})
        if status != 200 or not isinstance(body, dict) or not body.get("token"):
            raise BackupError(f"GuruRMM login failed (HTTP {status}): {body}", status=status)
        self._token = body["token"]

    def _auth(self) -> dict:
        if not self._token:
            self.login()
        return {"Authorization": f"Bearer {self._token}"}

    def list_agents(self) -> list[dict]:
        if self._agents is None:
            status, body = http_json("GET", f"{self.base}/api/agents", headers=self._auth())
            if status != 200 or not isinstance(body, list):
                raise BackupError(f"GET /api/agents failed (HTTP {status})", status=status)
            self._agents = body
        return self._agents

    def find_agents(self, client_substr: Optional[str] = None,
                    hostname_substr: Optional[str] = None) -> list[dict]:
        out = []
        for a in self.list_agents():
            if client_substr and client_substr.lower() not in (a.get("client_name") or "").lower():
                continue
            if hostname_substr and hostname_substr.lower() not in (a.get("hostname") or "").lower():
                continue
            out.append(a)
        return out

    def run_ps(self, agent_id: str, script: str, timeout_seconds: int = 90,
               poll_seconds: int = 4, max_polls: int = 30) -> dict:
        """Dispatch a PowerShell command to an agent and poll to completion."""
        status, body = http_json(
            "POST", f"{self.base}/api/agents/{agent_id}/command", headers=self._auth(),
            body={"command_type": "powershell", "command": script,
                  "timeout_seconds": timeout_seconds})
        if status not in (200, 201) or not isinstance(body, dict):
            raise BackupError(f"command dispatch failed (HTTP {status}): {body}", status=status)
        cid = body.get("command_id") or body.get("id")
        if not cid:
            raise BackupError(f"no command_id in dispatch response: {body}")
        terminal = {"completed", "failed", "cancelled", "interrupted", "timeout"}
        last: dict = {}
        for _ in range(max_polls):
            s, cb = http_json("GET", f"{self.base}/api/commands/{cid}", headers=self._auth())
            if s == 200 and isinstance(cb, dict):
                last = cb
                if (cb.get("status") or "") in terminal:
                    break
            time.sleep(poll_seconds)
        return last

    def detect_client(self, agent: dict, patterns: list[str], **kw) -> dict:
        """Run the detection script on one agent for the given product name-fragments."""
        pat_lits = ",".join("'" + p.replace("'", "''") + "'" for p in patterns)
        script = _DETECT_PS.replace("__PATTERNS__", pat_lits)
        res = self.run_ps(agent["id"], script, **kw)
        out = {"hostname": agent.get("hostname"), "client_name": agent.get("client_name"),
               "agent_id": agent.get("id"), "status": res.get("status"),
               "exit_code": res.get("exit_code"), "detected": None, "detail": None}
        stdout = (res.get("stdout") or "").strip()
        if stdout:
            try:
                parsed = json.loads(stdout)
                out["detected"] = bool(parsed.get("detected"))
                out["detail"] = parsed
            except json.JSONDecodeError:
                out["detail"] = {"_raw": stdout[:600]}
        return out


def eprint(*a: Any) -> None:
    print(*a, file=sys.stderr)