#!/usr/bin/env python3 """Datto Workplace REST API v1 client for the `datto-workplace` skill. Talks to ACG's live Datto Workplace tenant file API. Read-only: this is a FILE-ACCESS API (projects + files), NOT an admin/reporting API -- there are no user/device/member/team endpoints. It answers "what shared projects + files does this API principal see" for the GPS backup audit. Auth: HTTP Basic, username = api-key, password = api-secret (no token exchange; Basic is sent on every request). Credentials come from the SOPS vault. Base URL defaults to the team file-API host. The team path /6/api/v1 is normalized by the server to /api/v1; override with DATTO_WORKPLACE_URL if needed. IMPORTANT GOTCHA: GET /file/projects currently returns {"result": []} for this key -- the principal has no projects shared to it, so the file API alone CANNOT enumerate the team's backups. Handle empty gracefully. Because of that, RMM endpoint detection (see datto_workplace.py `audit --rmm`) is the PRIMARY source of "who backs up to Datto Workplace". """ from __future__ import annotations import os import sys import urllib.parse from pathlib import Path from typing import Any, Optional # Import the shared backup helpers from .claude/scripts. _SKILL_DIR = Path(__file__).resolve().parent.parent # .../skills/datto-workplace _SCRIPTS_DIR = _SKILL_DIR.parent.parent / "scripts" # .../.claude/scripts sys.path.insert(0, str(_SCRIPTS_DIR)) import backup_common as bc # noqa: E402 VAULT_ENTRY = "msp-tools/datto-workplace.sops.yaml" VAULT_FIELD_KEY = "credentials.api-key" VAULT_FIELD_SECRET = "credentials.api-secret" DEFAULT_URL = os.environ.get("DATTO_WORKPLACE_URL", "https://acg.workplace.datto.com/api/v1") class DattoWorkplaceClient: """Basic-auth GET client for the Datto Workplace file API. No token caching is needed -- HTTP Basic is sent per request. Credentials are read once from the vault on first use and reused for the process lifetime. """ def __init__(self, base: Optional[str] = None): self.base = (base or DEFAULT_URL).rstrip("/") self._basic: Optional[tuple[str, str]] = None # -- auth ------------------------------------------------------------------ def _creds(self) -> tuple[str, str]: if self._basic is None: key = bc.vault_get_field(VAULT_ENTRY, VAULT_FIELD_KEY) secret = bc.vault_get_field(VAULT_ENTRY, VAULT_FIELD_SECRET) self._basic = (key, secret) return self._basic def _get_raw(self, path: str, params: Optional[dict] = None) -> tuple[int, Any]: """Low-level GET returning (status, body); callers decide how to react.""" url = f"{self.base}{path}" if params: url += "?" + urllib.parse.urlencode(params) return bc.http_json("GET", url, basic=self._creds()) def _get(self, path: str, params: Optional[dict] = None) -> Any: status, body = self._get_raw(path, params) if status == 401: raise bc.BackupError( f"Datto Workplace auth failed (HTTP 401) for {path}: check api-key/api-secret", status=status) if status != 200: raise bc.BackupError( f"Datto Workplace GET {path} failed (HTTP {status}): {body}", status=status) return body # -- read methods ---------------------------------------------------------- def openapi(self) -> dict: """Fetch the OpenAPI spec (confirms reachability + exact endpoint list).""" body = self._get("/openapi.json") return body if isinstance(body, dict) else {"_raw": body} def endpoint_paths(self) -> list[str]: """Sorted list of paths the OpenAPI spec exposes.""" spec = self.openapi() paths = spec.get("paths") if isinstance(spec, dict) else None return sorted(paths.keys()) if isinstance(paths, dict) else [] def projects(self) -> list[dict]: """Top-level file projects visible to this API principal. Returns [] gracefully -- for this key the principal currently has no projects shared to it. The server is inconsistent here: some responses are 200 {"result": []}, others are 403 "Access denied" for the exact same Basic auth that authorizes /openapi.json seconds earlier. Both mean the same thing for the audit (no project-level file access), so a 403/401 on this endpoint is treated as empty rather than a hard error. A genuine credential failure surfaces on /openapi.json (status/openapi). """ status, body = self._get_raw("/file/projects") if status in (401, 403): return [] if status != 200: raise bc.BackupError( f"Datto Workplace GET /file/projects failed (HTTP {status}): {body}", status=status) return self._as_list(body) def files(self, parent_id: str) -> list[dict]: """List files/folders under a parent project or folder ID.""" body = self._get(f"/file/{parent_id}/files") return self._as_list(body) def file_info(self, file_id: str) -> dict: """Metadata for a single file/folder ID.""" body = self._get(f"/file/{file_id}") if isinstance(body, dict): return body.get("result") if isinstance(body.get("result"), dict) else body return {"_raw": body} def search(self, query: str, parent_id: Optional[str] = None) -> list[dict]: """Search files by name. Optional parent scope.""" params = {"query": query} if parent_id: params["parentID"] = parent_id body = self._get("/file/search", params) return self._as_list(body) @staticmethod def _as_list(body: Any) -> list[dict]: """Normalize the Datto file-API envelope to a list. Responses wrap payloads as {"result": [...]} (or a bare list on some endpoints); anything else yields []. """ if isinstance(body, dict): res = body.get("result") if isinstance(res, list): return res if isinstance(res, dict): return [res] return [] if isinstance(body, list): return body return [] def main() -> int: try: c = DattoWorkplaceClient() paths = c.endpoint_paths() projects = c.projects() print(f"[OK] Datto Workplace file API reachable at {c.base}; " f"{len(paths)} endpoints, {len(projects)} project(s) visible") return 0 except bc.BackupError as exc: print(f"[ERROR] {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())