166 lines
6.9 KiB
Python
166 lines
6.9 KiB
Python
#!/usr/bin/env python3
|
|
"""CLI for the `datto-workplace` skill -- read-only Datto Workplace backup
|
|
verification for the GPS audit.
|
|
|
|
Commands:
|
|
status file API reachable under Basic auth + endpoint/project count
|
|
projects [--json] list file-API projects (likely empty for this principal)
|
|
audit [--client NAME] [--rmm] [--json]
|
|
the GPS view: server-side projects (if any), optionally
|
|
cross-checked against GuruRMM endpoints (Datto Workplace
|
|
client installed + signed in?)
|
|
|
|
The Datto Workplace file API is FILE-ACCESS ONLY and this principal sees no shared
|
|
projects, so it CANNOT enumerate the team's backups on its own. RMM endpoint
|
|
detection (`audit --rmm`) is therefore the PRIMARY source: it detects the Datto
|
|
Workplace client on GuruRMM agents using precise patterns that do NOT false-match
|
|
Datto EDR / Datto RMM / Datto AV.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
from datto_client import DattoWorkplaceClient # noqa: E402
|
|
import backup_common as bc # noqa: E402
|
|
|
|
# Precise detection patterns. "Datto Workplace" catches the DisplayName of both the
|
|
# v10 client ("Datto Workplace v10.x") and legacy v8 ("Datto Workplace Desktop");
|
|
# "Workplace2" catches the v10 install path / process family. A bare "Datto" would
|
|
# false-match "Datto EDR Agent", Datto RMM, and Datto AV -- do NOT use it.
|
|
DETECT_PATTERNS = ["Datto Workplace", "Workplace2"]
|
|
|
|
|
|
def cmd_status(c: DattoWorkplaceClient, args) -> int:
|
|
paths = c.endpoint_paths()
|
|
projects = c.projects()
|
|
if args.json:
|
|
print(json.dumps({"base": c.base, "reachable": True,
|
|
"endpoints": paths, "project_count": len(projects)}, indent=2))
|
|
return 0
|
|
print(f"[OK] Datto Workplace file API reachable at {c.base} (Basic auth 200)")
|
|
print(f"[INFO] {len(paths)} endpoints exposed:")
|
|
for p in paths:
|
|
print(f" {p}")
|
|
print(f"[INFO] {len(projects)} project(s) visible to this API principal", end="")
|
|
if not projects:
|
|
print(" - file API cannot enumerate backups; use `audit --rmm`")
|
|
else:
|
|
print()
|
|
return 0
|
|
|
|
|
|
def cmd_projects(c: DattoWorkplaceClient, args) -> int:
|
|
projects = c.projects()
|
|
if args.json:
|
|
print(json.dumps(projects, indent=2))
|
|
return 0
|
|
if not projects:
|
|
print("[INFO] no file-API projects visible to this principal.")
|
|
print("[INFO] The api-key has no projects shared to it, so GET /file/projects "
|
|
"returns []. This is expected - the Datto Workplace file API is file-access "
|
|
"only. Use `audit --rmm` to verify backups via GuruRMM endpoint detection.")
|
|
return 0
|
|
print(f"{'NAME':40} {'ID':38} {'FILES':>8}")
|
|
for p in projects:
|
|
print(f"{str(p.get('name') or '')[:40]:40} "
|
|
f"{str(p.get('id') or p.get('projectID') or '')[:38]:38} "
|
|
f"{str(p.get('fileCount') or p.get('count') or '?'):>8}")
|
|
print(f"\n{len(projects)} project(s)")
|
|
return 0
|
|
|
|
|
|
def cmd_audit(c: DattoWorkplaceClient, args) -> int:
|
|
projects = c.projects()
|
|
result = {"backend": "datto-workplace", "base": c.base,
|
|
"file_api_projects": projects, "file_api_limited": not projects}
|
|
|
|
if args.rmm:
|
|
rmm = bc.RmmClient()
|
|
rmm.login()
|
|
checks = []
|
|
agents = rmm.find_agents(client_substr=args.client) if args.client else rmm.list_agents()
|
|
for a in agents:
|
|
det = rmm.detect_client(a, DETECT_PATTERNS)
|
|
if det.get("detected"):
|
|
checks.append(det)
|
|
result["rmm_endpoints_with_client"] = checks
|
|
result["rmm_agents_scanned"] = len(agents)
|
|
|
|
if args.json:
|
|
print(json.dumps(result, indent=2))
|
|
return 0
|
|
|
|
print(f"=== Datto Workplace backup audit - {c.base} ===")
|
|
if projects:
|
|
print(f"{'PROJECT':40} {'ID':38} {'FILES':>8}")
|
|
for p in projects:
|
|
print(f"{str(p.get('name') or '')[:40]:40} "
|
|
f"{str(p.get('id') or p.get('projectID') or '')[:38]:38} "
|
|
f"{str(p.get('fileCount') or p.get('count') or '?'):>8}")
|
|
print(f"\n{len(projects)} file-API project(s)")
|
|
else:
|
|
print("[INFO] file API shows 0 projects (principal has none shared) - "
|
|
"server-side enumeration not available; relying on RMM detection.")
|
|
|
|
if args.rmm:
|
|
eps = result.get("rmm_endpoints_with_client", [])
|
|
scanned = result.get("rmm_agents_scanned", 0)
|
|
scope = f"client~'{args.client}'" if args.client else "ALL agents"
|
|
print(f"\n--- GuruRMM endpoints running the Datto Workplace client "
|
|
f"({len(eps)} of {scanned} scanned, {scope}) ---")
|
|
for e in eps:
|
|
det = e.get("detail") or {}
|
|
installed = det.get("installed") or []
|
|
prod = "-"
|
|
if isinstance(installed, list) and installed:
|
|
first = installed[0]
|
|
if isinstance(first, dict):
|
|
prod = f"{first.get('name') or '?'} {first.get('version') or ''}".strip()
|
|
elif isinstance(installed, dict):
|
|
prod = f"{installed.get('name') or '?'} {installed.get('version') or ''}".strip()
|
|
print(f" {str(e.get('hostname') or '')[:25]:25} "
|
|
f"client={str(e.get('client_name') or ''):20} "
|
|
f"product={prod}")
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(prog="datto-workplace",
|
|
description="Datto Workplace backup verification (GPS audit)")
|
|
p.add_argument("--url", help="override file-API base URL (default team file-API host)")
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
|
|
s = sub.add_parser("status"); s.add_argument("--json", action="store_true")
|
|
pr = sub.add_parser("projects"); pr.add_argument("--json", action="store_true")
|
|
a = sub.add_parser("audit")
|
|
a.add_argument("--client", help="scope RMM detection to agents whose client_name matches")
|
|
a.add_argument("--rmm", action="store_true", help="cross-check GuruRMM endpoints (primary path)")
|
|
a.add_argument("--json", action="store_true")
|
|
return p
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
c = DattoWorkplaceClient(args.url)
|
|
handlers = {"status": cmd_status, "projects": cmd_projects, "audit": cmd_audit}
|
|
try:
|
|
return handlers[args.cmd](c, args)
|
|
except bc.BackupError as exc:
|
|
print(f"[ERROR] {exc}", file=sys.stderr)
|
|
# best-effort error logging per house rule
|
|
try:
|
|
root = bc.resolve_root()
|
|
bc.subprocess.run(["bash", str(root / ".claude/scripts/log-skill-error.sh"),
|
|
"datto-workplace", str(exc)[:200]], timeout=15)
|
|
except Exception:
|
|
pass
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|