260 lines
11 KiB
Python
260 lines
11 KiB
Python
#!/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.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, 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).
|
|
|
|
4xx/5xx are returned (not raised) so callers can inspect the error body.
|
|
Transport failures raise BackupError.
|
|
"""
|
|
hdrs = dict(headers or {})
|
|
data = None
|
|
if body is not None:
|
|
data = json.dumps(body).encode("utf-8")
|
|
hdrs.setdefault("Content-Type", "application/json")
|
|
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)
|