claudetools/.claude/skills/datto-workplace/scripts/datto_client.py

#!/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())