340 lines
13 KiB
Python
340 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""CLI for the datto-edr skill - Datto EDR (Infocyte HUNT) REST API.
|
|
|
|
Read subcommands run freely. Mutating subcommands (scan, isolate, run-extension)
|
|
refuse to run unless --confirm is passed; without it they print what they WOULD
|
|
do and exit non-zero.
|
|
|
|
Output: --json emits raw JSON; otherwise a readable table/summary.
|
|
|
|
Usage examples:
|
|
python edr.py status
|
|
python edr.py orgs # clients (Organizations)
|
|
python edr.py sites [--org <orgId>] # target groups
|
|
python edr.py agents [--org <orgId>] [--site <targetGroupId>]
|
|
python edr.py agent <agentId>
|
|
python edr.py detections [--org <orgId>] [--severity 3] [--days 7]
|
|
python edr.py detection <alertId>
|
|
python edr.py sweep [--org <orgId>] # per-client posture rollup
|
|
python edr.py deploy-cmd [--regkey <key>] # emit agent install one-liner
|
|
python edr.py extensions # list response/collection extensions
|
|
python edr.py scan --site <targetGroupId> --confirm
|
|
python edr.py scan --site <targetGroupId> --target <ip/host> --confirm
|
|
python edr.py isolate --site <tgId> --target <host> --extension-name "Host Isolation" --confirm
|
|
python edr.py raw GET Agents --filter '{"limit":1}'
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
from edr_client import DattoEDRClient, DattoEDRError, DEFAULT_INSTANCE
|
|
|
|
|
|
def _log_skill_error(skill, msg, context=""):
|
|
"""Soft-fail: append a functional-error entry to errorlog.md (never throws)."""
|
|
try:
|
|
root = os.environ.get("CLAUDETOOLS_ROOT") or os.path.abspath(
|
|
os.path.join(os.path.dirname(__file__), "..", "..", "..", "..")
|
|
)
|
|
h = os.path.join(root, ".claude", "scripts", "log-skill-error.sh")
|
|
if not os.path.exists(h):
|
|
return
|
|
a = ["bash", h, skill, msg]
|
|
if context:
|
|
a += ["--context", context]
|
|
subprocess.run(a, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
|
|
timeout=10)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
_EXPECTED_ERROR_MARKERS = (
|
|
"404", "not found", "no method to handle", "invalid value",
|
|
"required", "must be", "429", "too many requests", "unauthorized",
|
|
)
|
|
|
|
|
|
def _is_expected_error(msg: str) -> bool:
|
|
m = (msg or "").lower()
|
|
return any(marker in m for marker in _EXPECTED_ERROR_MARKERS)
|
|
|
|
|
|
def _should_log_error(command: str, msg: str) -> bool:
|
|
if os.environ.get("EDR_SUPPRESS_ERRORLOG"):
|
|
return False
|
|
if command == "raw":
|
|
return False
|
|
return not _is_expected_error(msg)
|
|
|
|
|
|
def _emit(obj, as_json: bool, table_fn=None) -> None:
|
|
if as_json or table_fn is None:
|
|
print(json.dumps(obj, indent=2, default=str))
|
|
else:
|
|
table_fn(obj)
|
|
|
|
|
|
def _trunc(s, n):
|
|
s = "" if s is None else str(s)
|
|
return s if len(s) <= n else s[: n - 1] + "…"
|
|
|
|
|
|
# --- table renderers ----------------------------------------------------------
|
|
def _t_status(s):
|
|
print(f"Datto EDR tenant: {s.get('instance')}")
|
|
print(f" Organizations (clients): {s.get('organizations')}")
|
|
print(f" Target groups (sites) : {s.get('targetGroups')}")
|
|
print(f" Agents : {s.get('agents')} "
|
|
f"({s.get('agentsActive')} active, {s.get('agentsIsolated')} isolated)")
|
|
print(f" Alerts (all time) : {s.get('alerts')}")
|
|
|
|
|
|
def _t_orgs(rows):
|
|
print(f"{'ORGANIZATION':<34} {'AGENTS':>7} {'ALERTS':>7} {'SITES':>6} ID")
|
|
for o in rows:
|
|
print(f"{_trunc(o.get('name'),34):<34} {o.get('agentCount',0):>7} "
|
|
f"{o.get('alertCount',0):>7} {o.get('locationCount',0):>6} {o.get('id')}")
|
|
print(f"\n{len(rows)} organizations")
|
|
|
|
|
|
def _t_sites(rows):
|
|
print(f"{'SITE (TARGET GROUP)':<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")
|
|
|
|
|
|
def _agent_os(a):
|
|
for k, label in (("osWindows", "Windows"), ("osLinux", "Linux"),
|
|
("osOsx", "macOS"), ("osOther", "Other")):
|
|
if a.get(k):
|
|
return label
|
|
return a.get("os") or "?"
|
|
|
|
|
|
def _t_agents(rows):
|
|
print(f"{'HOSTNAME':<26} {'OS':<8} {'ONLINE':<7} {'ISO':<4} {'AV':<4} "
|
|
f"{'VERSION':<10} 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')}")
|
|
print(f"\n{len(rows)} agents")
|
|
|
|
|
|
_SEV = {0: "info", 1: "low", 2: "medium", 3: "high", 4: "critical"}
|
|
|
|
|
|
def _t_detections(rows):
|
|
print(f"{'WHEN':<20} {'SEV':<8} {'HOST':<22} {'ORG':<20} NAME")
|
|
for a in rows:
|
|
sev = _SEV.get(a.get("severity"), str(a.get("severity")))
|
|
print(f"{_trunc(a.get('createdOn'),20):<20} {sev:<8} "
|
|
f"{_trunc(a.get('hostname'),22):<22} "
|
|
f"{_trunc(a.get('organizationName'),20):<20} "
|
|
f"{_trunc(a.get('name') or a.get('description'),40)}")
|
|
print(f"\n{len(rows)} detections")
|
|
|
|
|
|
def _t_sweep(s):
|
|
rows = s.get("clients", [])
|
|
print(f"{'CLIENT':<34} {'AGENTS':>7} {'ALERTS(7d)':>11} {'ALERTS(all)':>12}")
|
|
for r in rows:
|
|
print(f"{_trunc(r.get('organization'),34):<34} {r.get('agents',0):>7} "
|
|
f"{r.get('alertsLast7d',0):>11} {r.get('alertsTotal',0):>12}")
|
|
print(f"\n{len(rows)} clients (sorted by 7-day detection volume)")
|
|
|
|
|
|
# --- 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
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(prog="edr.py", description="Datto EDR (Infocyte HUNT) CLI")
|
|
p.add_argument("--json", action="store_true", help="emit raw JSON")
|
|
sub = p.add_subparsers(dest="command", required=True)
|
|
|
|
sub.add_parser("status", help="tenant rollup")
|
|
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.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("--limit", type=int, default=500)
|
|
|
|
sp = sub.add_parser("agent", help="agent detail")
|
|
sp.add_argument("id")
|
|
|
|
sp = sub.add_parser("detections", help="list alerts")
|
|
sp.add_argument("--org")
|
|
sp.add_argument("--site")
|
|
sp.add_argument("--severity", type=int, help="0 info..4 critical")
|
|
sp.add_argument("--days", type=int)
|
|
sp.add_argument("--limit", type=int, default=200)
|
|
|
|
sp = sub.add_parser("detection", help="alert detail")
|
|
sp.add_argument("id")
|
|
|
|
sp = sub.add_parser("sweep", help="per-client posture rollup")
|
|
sp.add_argument("--org")
|
|
|
|
sub.add_parser("extensions", help="list agent extensions")
|
|
|
|
sp = sub.add_parser("deploy-cmd", help="emit the agent install one-liner")
|
|
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", 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", required=True, help="ip/host")
|
|
sp.add_argument("--extension-id")
|
|
sp.add_argument("--extension-name", default="Host Isolation")
|
|
sp.add_argument("--confirm", action="store_true")
|
|
|
|
sp = sub.add_parser("raw", help="call any endpoint")
|
|
sp.add_argument("method", help="GET/POST/PUT/DELETE")
|
|
sp.add_argument("path", help="e.g. Agents or targets/{id}/scan")
|
|
sp.add_argument("--filter", help="LoopBack filter JSON (GET)")
|
|
sp.add_argument("--data", help="request body JSON")
|
|
sp.add_argument("--confirm", action="store_true",
|
|
help="required for non-GET methods")
|
|
return p
|
|
|
|
|
|
def _gate(args, what):
|
|
"""Print the would-do and return False if --confirm missing."""
|
|
if getattr(args, "confirm", False):
|
|
return True
|
|
print(f"[DRY-RUN] would {what}\n[BLOCKED] re-run with --confirm to execute.",
|
|
file=sys.stderr)
|
|
return False
|
|
|
|
|
|
def main(argv=None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
j = args.json
|
|
client = DattoEDRClient()
|
|
cmd = args.command
|
|
|
|
try:
|
|
if cmd == "status":
|
|
_emit(client.get_status(), j, _t_status)
|
|
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)
|
|
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)
|
|
_emit(rows, j, _t_agents)
|
|
elif cmd == "agent":
|
|
_emit(client.get_agent(args.id), j)
|
|
elif cmd == "detections":
|
|
_emit(client.list_alerts(org_id=args.org, target_group_id=args.site,
|
|
severity=args.severity, days=args.days,
|
|
limit=args.limit), j, _t_detections)
|
|
elif cmd == "detection":
|
|
_emit(client.get_alert(args.id), j)
|
|
elif cmd == "sweep":
|
|
_emit(client.sweep(org_id=args.org), j, _t_sweep)
|
|
elif cmd == "extensions":
|
|
_emit(client.list_extensions(), j)
|
|
elif cmd == "deploy-cmd":
|
|
regkey = args.regkey
|
|
if not regkey:
|
|
keys = client.list_agent_keys()
|
|
if isinstance(keys, list) and keys:
|
|
regkey = keys[0].get("key") or keys[0].get("id")
|
|
_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}"):
|
|
return 2
|
|
if args.target:
|
|
res = client.scan_single_target(args.site, args.target)
|
|
else:
|
|
res = client.scan_target_group(args.site)
|
|
_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})"):
|
|
return 2
|
|
res = client.run_response_extension(
|
|
args.site, args.target,
|
|
extension_id=args.extension_id,
|
|
extension_name=None if args.extension_id else args.extension_name)
|
|
_emit(res, j)
|
|
elif cmd == "raw":
|
|
method = args.method.upper()
|
|
if method != "GET" and not args.confirm:
|
|
print(f"[BLOCKED] {method} requires --confirm.", file=sys.stderr)
|
|
return 2
|
|
filt = json.loads(args.filter) if args.filter else None
|
|
body = json.loads(args.data) if args.data else None
|
|
_emit(client.raw(method, args.path, filt=filt, body=body), j)
|
|
else:
|
|
print(f"[ERROR] unknown command {cmd}", file=sys.stderr)
|
|
return 1
|
|
except DattoEDRError as exc:
|
|
msg = str(exc)
|
|
print(f"[ERROR] {msg}", file=sys.stderr)
|
|
if _should_log_error(cmd, msg):
|
|
_log_skill_error("datto-edr", msg, context=f"cmd={cmd}")
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _deploy_payload(client, regkey):
|
|
instance = client.api_base_url.replace("https://", "").replace("/api", "")
|
|
cname = instance.split(".")[0]
|
|
keypart = f" -RegKey {regkey}" if regkey else ""
|
|
oneliner = (
|
|
"[System.Net.ServicePointManager]::SecurityProtocol = "
|
|
"[Enum]::ToObject([System.Net.SecurityProtocolType], 3072); "
|
|
"(new-object Net.WebClient).DownloadString("
|
|
"\"https://raw.githubusercontent.com/Infocyte/PowershellTools/master/"
|
|
"AgentDeployment/install_huntagent.ps1\") | iex; "
|
|
f"Install-EDR -InstanceName {cname}{keypart}"
|
|
)
|
|
return {
|
|
"instance": cname,
|
|
"regkey": regkey or "(none found - agent will register but await approval)",
|
|
"powershell_oneliner": oneliner,
|
|
"note": "Run on the target via GuruRMM script-execution or any remote-exec "
|
|
"channel. RegKey auto-approves + adds to the default target group.",
|
|
}
|
|
|
|
|
|
def _t_deploy(p):
|
|
print(f"Instance : {p['instance']}")
|
|
print(f"RegKey : {p['regkey']}")
|
|
print(f"\nPowerShell one-liner (run on target, admin):\n\n{p['powershell_oneliner']}\n")
|
|
print(f"[note] {p['note']}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|