azcomputerguru/claudetools
1
0
Fork 0
Code Issues 59 Pull Requests Actions Packages Projects Releases Wiki Activity

skills: add datto-edr (Datto EDR / Infocyte HUNT) + syncro-rmm memory

Browse Source 
New /datto-edr skill — standalone CLI for the Datto EDR REST API (azcomp4587,
rebranded Infocyte HUNT, raw-token LoopBack). Live-verified reads across the whole
MSP fleet: orgs/sites/agents/detections/sweep + deploy-cmd. Scan/isolate gated
behind --confirm (shape-verified, not run against prod). Token vaulted at
msp-tools/datto-edr.sops.yaml.

Also: reference_syncro_rmm_api_gui_only memory (Syncro RMM policy mgmt is
GUI-only) and the guru-rmm submodule pointer bump (Feature 6 EDR research).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Howard Enos
2026-06-25 12:39:42 -07:00
parent e61b39b5c8
commit bd1e84d32c
7 changed files with 377 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
.claude/skills/datto-edr/scripts/edr.py
View File

@@ -102,11 +102,19 @@ def _t_orgs(rows):
def _t_sites(rows):
print(f"{'SITE (TARGET GROUP)':<34} {'AGENTS':>7} {'ACTIVE':>7} {'ALERTS':>7} ID")
print(f"{'SITE (LOCATION)':<34} {'AGENTS':>7} {'ACTIVE':>7} {'ALERTS':>7} ID")
for t in rows:
print(f"{_trunc(t.get('name'),34):<34} {t.get('agentCount',0):>7} "
f"{t.get('activeAgentCount',0):>7} {t.get('alertCount',0):>7} {t.get('id')}")
print(f"\n{len(rows)} target groups")
print(f"\n{len(rows)} locations")
def _t_scan_targets(rows):
print(f"{'SCAN TARGET GROUP':<34} {'AGENTS':>7} {'ACTIVE':>7} {'LAST SCAN':<20} ID")
for t in rows:
print(f"{_trunc(t.get('name'),34):<34} {t.get('agentCount',0):>7} "
f"{t.get('activeAgentCount',0):>7} {_trunc(t.get('lastScannedOn'),20):<20} {t.get('id')}")
print(f"\n{len(rows)} scan target groups")
def _agent_os(a):
@@ -119,13 +127,13 @@ def _agent_os(a):
def _t_agents(rows):
print(f"{'HOSTNAME':<26} {'OS':<8} {'ONLINE':<7} {'ISO':<4} {'AV':<4} "
f"{'VERSION':<10} ID")
f"{'VERSION':<13} ID")
for a in rows:
print(f"{_trunc(a.get('hostname') or a.get('name'),26):<26} "
f"{_agent_os(a):<8} {('yes' if a.get('active') else 'no'):<7} "
f"{('ISO' if a.get('isolated') else '-'):<4} "
f"{('on' if a.get('dattoAvEnabled') else '-'):<4} "
f"{_trunc(a.get('version'),10):<10} {a.get('id')}")
f"{_trunc(a.get('version'),13):<13} {a.get('id')}")
print(f"\n{len(rows)} agents")
@@ -154,11 +162,11 @@ def _t_sweep(s):
# --- org -> target group resolution -------------------------------------------
def _agents_for_org(client, org_id, limit):
"""Agents across all target groups in an org (Agents carry deviceGroupId)."""
out = []
for tg in client.list_targets(org_id=org_id):
out.extend(client.list_agents(target_group_id=tg.get("id"), limit=limit))
return out
"""Agents across all sites (Locations) in an org. Agent.locationId -> Location."""
loc_ids = [l.get("id") for l in client.list_locations(org_id=org_id) if l.get("id")]
if not loc_ids:
return []
return client.list_agents(location_ids=loc_ids, limit=limit)
def build_parser() -> argparse.ArgumentParser:
@@ -170,12 +178,15 @@ def build_parser() -> argparse.ArgumentParser:
sub.add_parser("orgs", help="list organizations (clients)")
sub.add_parser("clients", help="alias for orgs")
sp = sub.add_parser("sites", help="list target groups")
sp = sub.add_parser("sites", help="list sites (Locations)")
sp.add_argument("--org", help="filter by organizationId")
sp = sub.add_parser("scan-targets", help="list scan target groups")
sp.add_argument("--org", help="filter by organizationId")
sp = sub.add_parser("agents", help="list agents")
sp.add_argument("--org")
sp.add_argument("--site", help="target group id")
sp.add_argument("--site", help="location id (a single site)")
sp.add_argument("--limit", type=int, default=500)
sp = sub.add_parser("agent", help="agent detail")
@@ -200,12 +211,13 @@ def build_parser() -> argparse.ArgumentParser:
sp.add_argument("--regkey", help="registration key (else pulled from /agentKeys)")
sp = sub.add_parser("scan", help="trigger a scan (gated)")
sp.add_argument("--site", required=True, help="target group id")
sp.add_argument("--target-group", required=True,
help="scan target group id (from scan-targets)")
sp.add_argument("--target", help="single ip/host (else whole group)")
sp.add_argument("--confirm", action="store_true")
sp = sub.add_parser("isolate", help="invoke a response extension e.g. host isolation (gated)")
sp.add_argument("--site", required=True, help="target group id")
sp.add_argument("--target-group", required=True, help="scan target group id")
sp.add_argument("--target", required=True, help="ip/host")
sp.add_argument("--extension-id")
sp.add_argument("--extension-name", default="Host Isolation")
@@ -242,12 +254,14 @@ def main(argv=None) -> int:
elif cmd in ("orgs", "clients"):
_emit(client.list_organizations(), j, _t_orgs)
elif cmd == "sites":
_emit(client.list_targets(org_id=args.org), j, _t_sites)
_emit(client.list_locations(org_id=args.org), j, _t_sites)
elif cmd == "scan-targets":
_emit(client.list_targets(org_id=args.org), j, _t_scan_targets)
elif cmd == "agents":
if args.org and not args.site:
rows = _agents_for_org(client, args.org, args.limit)
else:
rows = client.list_agents(target_group_id=args.site, limit=args.limit)
rows = client.list_agents(location_id=args.site, limit=args.limit)
_emit(rows, j, _t_agents)
elif cmd == "agent":
_emit(client.get_agent(args.id), j)
@@ -270,20 +284,20 @@ def main(argv=None) -> int:
_emit(_deploy_payload(client, regkey), j, _t_deploy)
elif cmd == "scan":
tgt = f"target {args.target}" if args.target else "the whole target group"
if not _gate(args, f"scan {tgt} in site {args.site}"):
if not _gate(args, f"scan {tgt} in target group {args.target_group}"):
return 2
if args.target:
res = client.scan_single_target(args.site, args.target)
res = client.scan_single_target(args.target_group, args.target)
else:
res = client.scan_target_group(args.site)
res = client.scan_target_group(args.target_group)
_emit(res, j)
elif cmd == "isolate":
if not _gate(args, f"run response extension "
f"'{args.extension_name or args.extension_id}' "
f"on {args.target} (site {args.site})"):
f"on {args.target} (target group {args.target_group})"):
return 2
res = client.run_response_extension(
args.site, args.target,
args.target_group, args.target,
extension_id=args.extension_id,
extension_name=None if args.extension_id else args.extension_name)
_emit(res, j)

33
.claude/skills/datto-edr/scripts/edr_client.py
View File

@@ -228,9 +228,23 @@ class DattoEDRClient:
"alertCount": True, "locationCount": True},
}) or []
def list_locations(self, org_id: Optional[str] = None, limit: int = 500) -> list[dict]:
"""List Locations (= client sites). VERIFIED LIVE. This is the per-client
grouping that agents belong to (Agent.locationId -> Location.id), and a
Location carries organizationId. Org -> Locations -> Agents is the
inventory hierarchy."""
filt: dict = {"limit": limit, "order": "name ASC",
"fields": {"id": True, "name": True, "organizationId": True,
"agentCount": True, "activeAgentCount": True,
"alertCount": True, "lastScannedOn": True}}
if org_id:
filt["where"] = {"organizationId": org_id}
return self._get("Locations", filt) or []
def list_targets(self, org_id: Optional[str] = None, limit: int = 500) -> list[dict]:
"""List Targets (= target groups / sites). VERIFIED LIVE.
Each Target belongs to an Organization via organizationId."""
"""List Targets (= SCAN target groups). VERIFIED LIVE. Distinct from
Locations: a Target is the scannable unit (POST targets/{id}/scan). Each
Target belongs to an Organization via organizationId."""
filt: dict = {"limit": limit, "order": "name ASC",
"fields": {"id": True, "name": True, "organizationId": True,
"agentCount": True, "activeAgentCount": True,
@@ -239,16 +253,17 @@ class DattoEDRClient:
filt["where"] = {"organizationId": org_id}
return self._get("Targets", filt) or []
def list_agents(self, org_id: Optional[str] = None,
target_group_id: Optional[str] = None,
def list_agents(self, location_id: Optional[str] = None,
location_ids: Optional[list[str]] = None,
limit: int = 500) -> list[dict]:
"""List Agents (endpoints). VERIFIED LIVE.
Filter by deviceGroupId (target group). Org filtering is done by the CLI
via the target-group->org map since Agents carry deviceGroupId, not orgId."""
"""List Agents (endpoints). VERIFIED LIVE. Filter by locationId (a single
site) or location_ids (an org's sites, via LoopBack `inq`)."""
filt: dict = {"limit": limit, "order": "hostname ASC"}
where: dict = {}
if target_group_id:
where["deviceGroupId"] = target_group_id
if location_id:
where["locationId"] = location_id
elif location_ids:
where["locationId"] = {"inq": location_ids}
if where:
filt["where"] = where
return self._get("Agents", filt) or []

51
.claude/skills/datto-edr/scripts/selftest.py Normal file
View File

@@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""Self-test for the datto-edr skill. Read-only; never mutates the tenant.
Confirms: token loads from vault, transport works, and the core read endpoints
return live data. Exits non-zero on any failure. Set EDR_SUPPRESS_ERRORLOG=1 so
expected probe errors don't spam errorlog.md.
"""
from __future__ import annotations
import os
import sys
os.environ.setdefault("EDR_SUPPRESS_ERRORLOG", "1")
from edr_client import DattoEDRClient, DattoEDRError
def main() -> int:
checks: list[tuple[str, bool, str]] = []
try:
client = DattoEDRClient()
_ = client.api_token
checks.append(("vault token load", True, ""))
except DattoEDRError as exc:
print(f"[FAIL] token load: {exc}", file=sys.stderr)
return 1
def check(name, fn):
try:
val = fn()
checks.append((name, True, str(val)))
except Exception as exc: # noqa: BLE001 - selftest summarizes failures
checks.append((name, False, str(exc)))
check("status rollup", lambda: client.get_status()["agents"])
check("list organizations", lambda: f"{len(client.list_organizations())} orgs")
check("list locations", lambda: f"{len(client.list_locations())} sites")
check("list targets", lambda: f"{len(client.list_targets())} scan groups")
check("list agents (limit 5)", lambda: f"{len(client.list_agents(limit=5))} agents")
check("list alerts (limit 5)", lambda: f"{len(client.list_alerts(limit=5))} alerts")
ok = True
for name, passed, detail in checks:
marker = "[OK]" if passed else "[FAIL]"
print(f"{marker} {name}{(' -> ' + detail) if detail else ''}")
ok = ok and passed
return 0 if ok else 1
if __name__ == "__main__":
raise SystemExit(main())