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

datto-edr: fix scan to verified Agents/scan endpoint + harden

Browse Source 
- Scan now uses POST Agents/scan with AND-wrapped where {and:[{id:[...]}]}
  (the Infocyte targets/{id}/scan routes are dead/404; bare {id:{inq}} returns
  HTTP 412 ambiguous-column). Verified live: single-agent scan -> 'Scanning 1 host'.
- scan/isolate REQUIRE explicit --agent ids; empty list refused (tenant-wide footgun).
- isolate rides Agents/scan with the Host Isolation extension in options.extensions;
  resolves --extension-name -> id via /Extensions.
- New subcommands: tasks, task, cancel, create-group, mint-key.
- deploy-cmd emits full -URL (not -InstanceName; cname 'azcomp4587' trips the
  install script's .com regex and leaves --url empty).
- Docs (SKILL.md + api-reference.md) rewritten to the verified endpoints + footguns.

Lifecycle verified end-to-end on RMM-TEST-MACHINE (create-group/mint-key/install/
register/scan/cancel).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Howard Enos
2026-06-25 13:55:22 -07:00
parent 778e12d83e
commit 1e80fb24db
4 changed files with 263 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -242,9 +242,10 @@ class DattoEDRClient:
return self._get("Locations", filt) or []
def list_targets(self, org_id: Optional[str] = None, limit: int = 500) -> list[dict]:
"""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."""
"""List Targets (= scan groups). VERIFIED LIVE. A grouping concept that
belongs to an Organization via organizationId (often aliases a Location
id). NOTE: scanning is per-AGENT via Agents/scan - the legacy
targets/{id}/scan route is DEAD (404) on this tenant."""
filt: dict = {"limit": limit, "order": "name ASC",
"fields": {"id": True, "name": True, "organizationId": True,
"agentCount": True, "activeAgentCount": True,
@@ -324,64 +325,98 @@ class DattoEDRClient:
# ======================================================================
# MUTATING METHODS (caller MUST gate behind --confirm)
# ----------------------------------------------------------------------
# Shapes derived from the KaseyaDEDR InfocyteHUNTAPI module (scan.ps1):
# scan a target group : POST targets/{targetGroupId}/scan {options, where}
# scan a single target: POST targets/scan {target, targetGroup, options}
# response/isolate : POST targets/scan with a response extension
# NOT executed against the live prod tenant during build (would scan/isolate
# real client machines). Treat as SHAPE-VERIFIED, RUN-UNVERIFIED until first
# deliberate use.
# The Infocyte-module scan routes (targets/{id}/scan, targets/scan, scans)
# are DEAD (404) on Datto EDR. The live mechanism, read verbatim from the
# console JS bundle, is:
# POST agents/scan {where:<loopback where>, options:<scan opts>, taskName}
# where `where` SELECTS the agents. An ABSENT/empty `where` scans the ENTIRE
# tenant (the 156-host footgun) - so scan_agents REFUSES an empty id list.
# Response extensions (isolate/kill/quarantine) ride the same call via
# options.extensions. (Verified live 2026-06-25.)
# ======================================================================
DEFAULT_SCAN_OPTIONS = {
"process": True, "module": True, "driver": True, "memory": True,
"account": True, "artifact": True, "autostart": True, "application": True,
"installed": True, "hook": False, "network": False, "events": True,
}
def scan_agents(self, agent_ids: list[str],
options: Optional[dict] = None,
task_name: str = "Scan - EDR") -> Any:
"""Trigger an EDR scan on specific agents (POST agents/scan).
def scan_target_group(self, target_group_id: str,
options: Optional[dict] = None,
where: Optional[dict] = None) -> Any:
"""Trigger a scan across a target group (POST targets/{id}/scan).
Targets by Agent.id via a LoopBack where filter:
{"where":{"id":{"inq":[ids]}}, "options":{}, "taskName":"Scan - EDR"}.
REFUSES an empty agent_ids list - an empty where scans the whole tenant.
`options` may carry EDR forensic toggles + extensions; empty {} is valid.
STATE-CHANGING - gate behind --confirm at the call site."""
body: dict = {"options": options or self.DEFAULT_SCAN_OPTIONS}
if where is not None:
body["where"] = where
return self._request("POST", f"targets/{target_group_id}/scan", body=body)
def scan_single_target(self, target_group_id: str, target: str,
options: Optional[dict] = None) -> Any:
"""Scan a single on-demand target (POST targets/scan).
STATE-CHANGING - gate behind --confirm at the call site."""
body: dict = {
"target": target,
"targetGroup": {"id": target_group_id},
"options": options or self.DEFAULT_SCAN_OPTIONS,
ids = [i for i in (agent_ids or []) if i]
if not ids:
raise DattoEDRError(
"scan_agents refused: empty agent list. An absent `where` scans "
"the ENTIRE tenant (156-host footgun) - pass explicit agent id(s)."
)
# AND-wrapped form is REQUIRED: a bare {id:{inq:[...]}} returns HTTP 412
# "column reference id is ambiguous" (the backend joins tables). The
# console builds {and:[{id:[...]}]}; verified live -> "Scanning 1 host".
body = {
"where": {"and": [{"id": ids}]},
"options": options or {},
"taskName": task_name,
}
return self._request("POST", "targets/scan", body=body)
return self._request("POST", "Agents/scan", body=body)
def list_extensions(self, limit: int = 200) -> Any:
"""List agent extensions (response + collection). Read.
Response extensions (e.g. host isolation) are invoked via a scan body."""
Response extensions (e.g. Host Isolation) are invoked via scan options."""
return self._get("Extensions", {"limit": limit}) or []
def run_response_extension(self, target_group_id: str, target: str,
extension_id: Optional[str] = None,
extension_name: Optional[str] = None) -> Any:
"""Invoke a response extension on a target (isolate/kill/quarantine).
Per the module, a response runs as a scan with an extension in options.
UNVERIFIED shape - confirm the extension id/name from list_extensions().
def run_extension_on_agents(self, agent_ids: list[str], extension_id: str,
task_name: str = "Response") -> Any:
"""Invoke a response extension (isolate/kill/quarantine) on specific
agents - rides POST agents/scan with the extension in options.extensions.
Same empty-list guard as scan_agents. Confirm the extension id via
list_extensions() (e.g. 'Host Isolation [Win/Linux]').
STATE-CHANGING - gate behind --confirm at the call site."""
ext: dict = {}
if extension_id:
ext["id"] = extension_id
if extension_name:
ext["name"] = extension_name
ids = [i for i in (agent_ids or []) if i]
if not ids:
raise DattoEDRError(
"run_extension_on_agents refused: empty agent list (would target "
"the whole tenant). Pass explicit agent id(s)."
)
if not extension_id:
raise DattoEDRError("run_extension_on_agents requires an extension_id.")
body = {
"target": target,
"targetGroup": {"id": target_group_id},
"options": {"extensions": [ext]},
"where": {"and": [{"id": ids}]}, # AND-wrapped (see scan_agents)
"options": {"extensions": [{"id": extension_id}]},
"taskName": task_name,
}
return self._request("POST", "targets/scan", body=body)
return self._request("POST", "Agents/scan", body=body)
# -- task status / control --------------------------------------------------
def list_tasks(self, type_: Optional[str] = None, limit: int = 20) -> list[dict]:
"""List userTasks (scan/analysis jobs), newest first. Read."""
filt: dict = {"limit": limit, "order": "createdOn DESC"}
if type_:
filt["where"] = {"type": type_}
return self._get("userTasks", filt) or []
def get_task(self, task_id: str) -> dict:
"""Get one userTask (scan job) detail. Read."""
return self._get(f"userTasks/{task_id}") or {}
def cancel_task(self, task_id: str) -> Any:
"""Cancel a running userTask, e.g. a scan (POST userTasks/{id}/cancel ->
204). STATE-CHANGING - gate behind --confirm at the call site."""
return self._request("POST", f"userTasks/{task_id}/cancel")
# -- provisioning -----------------------------------------------------------
def create_group(self, name: str, organization_id: str) -> Any:
"""Create an EDR target group under an org (POST Targets).
STATE-CHANGING - gate behind --confirm at the call site."""
return self._request("POST", "Targets",
body={"name": name, "organizationId": organization_id})
def mint_registration_key(self, target_id: str, key: str) -> Any:
"""Mint an agent registration key tied to a target group (POST agentKeys).
The key string `id` is CALLER-SUPPLIED (10-char); an absent id 500s.
STATE-CHANGING - gate behind --confirm at the call site."""
return self._request("POST", "agentKeys",
body={"id": key, "targetId": target_id})
# ======================================================================
# POWER TOOL