Files
claudetools/.claude/skills/bitdefender/scripts/gz.py
Howard Enos db6aa3683f fix(bitdefender): all-clients sweep, quarantine path, EDR controls, self-test
Several bugs found and fixed during live testing against the ACG GravityZone
tenant:
- security_sweep_all_clients: iterate each company (the companies container is
  not a valid endpoint parent; passing it 400'd the whole sweep)
- list_quarantine: use service-scoped path quarantine/computers with companyId
  (bare quarantine module 404'd; param is companyId not parentId)
- rename GZEndpointSummary.detection_active -> threat_detected with corrected
  semantics (True = active threat, tracks with infected; not an engine-on flag)
- status: readable sectioned table renderer for the nested apiKey/license dict
- portable CLAUDETOOLS_ROOT resolution (derive from file path, not a Windows
  literal) so it works on the Mac/Linux fleet

Adds scripts/selftest.py: a 29-check read-only harness (all passing) covering
every read command, --json, error exit codes, and destructive-action gating.
EDR/incident commands (blocklist, isolate/unisolate, blocklist-add/remove) and
raw destructive-method gating are included from this session's work.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-05-30 07:29:55 -07:00

564 lines
21 KiB
Python

#!/usr/bin/env python3
"""CLI for the bitdefender skill — GravityZone Cloud Public API.
Read-only subcommands run freely. Destructive subcommands (delete-endpoint,
delete-package, delete-group) 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 gz.py status
python gz.py companies
python gz.py endpoints --company 5c4280716c0318f3478b456e
python gz.py endpoint <endpointId>
python gz.py sweep --company <id>
python gz.py policies
python gz.py policy <policyId>
python gz.py packages
python gz.py quarantine --company <id>
python gz.py inventory --refresh
python gz.py create-package --name "Win Default" --company <id>
python gz.py install-links --package "Win Default" --company <id>
python gz.py scan --targets <id1> <id2> --type 2 --name "Full scan"
python gz.py move --endpoints <id1> <id2> --group <groupId>
python gz.py make-group --name "New Group" --parent <parentId>
python gz.py delete-endpoint <id> --confirm
python gz.py blocklist --company <id>
python gz.py incidents --company <id>
python gz.py isolate --endpoints <id1> <id2> --confirm
python gz.py unisolate --endpoints <id1> <id2> --confirm
python gz.py blocklist-add --company <id> --hashes <h1> <h2> --confirm
python gz.py blocklist-remove --id <hashItemId> --confirm
python gz.py raw --module network --method getEndpointsList --params '{"page":1}'
"""
from __future__ import annotations
import argparse
import dataclasses
import json
import sys
from gz_client import GravityZoneClient, GravityZoneError, GZEndpointSummary
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=_json_default))
else:
table_fn(obj)
def _json_default(o):
if dataclasses.is_dataclass(o):
return dataclasses.asdict(o)
return str(o)
# --- table renderers ----------------------------------------------------------
def _print_kv(d: dict) -> None:
for k, v in d.items():
print(f" {k}: {v}")
def _print_status(status: dict) -> None:
"""Render the nested API-status dict as readable sections.
get_api_status() returns {"apiKey": {...}, "license": {...}} and possibly
"_licenseWarning". Print a [section] header per top-level key, then indented
key: value lines for each sub-field. Non-dict top-level values (e.g. the
license warning string) print as a single indented line under their header.
"""
for section, body in status.items():
print(f"[{section}]")
if isinstance(body, dict):
for k, v in body.items():
print(f" {k}: {v}")
else:
print(f" {body}")
def _print_company_table(data: dict) -> None:
items = data.get("items", [])
print(f"Companies: {data.get('total', len(items))}")
for c in items:
print(f" {c.get('id','?'):26} {c.get('name','')}")
def _print_endpoint_table(data: dict) -> None:
items = data.get("items", [])
print(f"Endpoints: {data.get('total', len(items))}")
for e in items:
print(f" {e.get('id','?'):26} {e.get('name',''):30} "
f"{e.get('operatingSystemVersion', e.get('os',''))}")
def _print_sweep_table(summaries: list) -> None:
print(f"Endpoints swept: {len(summaries)}")
print(f" {'STATUS':10} {'NAME':30} {'AGENT':14} {'LAST SEEN'}")
for s in summaries:
flags = []
if s.infected:
flags.append("INFECTED")
if s.signature_outdated:
flags.append("SIG-OLD")
if s.product_outdated:
flags.append("PROD-OLD")
status = ",".join(flags) if flags else "OK"
print(f" {status:10} {s.name[:30]:30} {str(s.agent_version or '-'):14} "
f"{s.last_seen or '-'}")
def _print_policy_table(data: dict) -> None:
items = data.get("items", [])
print(f"Policies: {data.get('total', len(items))}")
for p in items:
print(f" {p.get('id','?'):26} {p.get('name','')}")
def _print_package_table(data: dict) -> None:
items = data.get("items", [])
print(f"Packages: {data.get('total', len(items))}")
for p in items:
print(f" {str(p.get('id','?')):26} {p.get('name','')}")
def _print_quarantine_table(data: dict) -> None:
items = data.get("items", [])
print(f"Quarantine items: {data.get('total', len(items))}")
for q in items:
print(f" {q.get('threatName','?'):30} {q.get('endpointName','')} "
f"{q.get('detectionTime','')}")
def _print_blocklist_table(data: dict) -> None:
items = data.get("items", [])
print(f"Blocklist items: {data.get('total', len(items))}")
print(f" {'ID':26} {'HTYPE':6} {'HASH':66} SOURCE-INFO")
for b in items:
print(f" {str(b.get('id','?')):26} {str(b.get('hashType','?')):6} "
f"{str(b.get('hash','')):66} {b.get('sourceInfo','')}")
def _print_incidents_table(data: dict) -> None:
items = data.get("items", [])
print(f"Incidents: {data.get('total', len(items))}")
for i in items:
print(f" {str(i.get('id','?')):26} {i.get('name', i.get('title','')):40} "
f"{i.get('severity', i.get('status',''))}")
def _print_inventory_table(cache: dict) -> None:
print(f"Inventory cached_at: {cache.get('fetched_at')}")
print(f" companies: {len(cache.get('companies', {}))}")
print(f" endpoints: {len(cache.get('endpoints', {}))}")
print(f" policies: {len(cache.get('policies', {}))}")
print(f" packages: {len(cache.get('packages', []))}")
print(f" groups: {len(cache.get('groups', {}))}")
# --- command handlers ---------------------------------------------------------
def cmd_status(client, args):
_emit(client.get_api_status(), args.json, _print_status)
def cmd_companies(client, args):
_emit(client.list_companies(), args.json, _print_company_table)
def cmd_endpoints(client, args):
_emit(client.list_endpoints(args.company, per_page=args.per_page),
args.json, _print_endpoint_table)
def cmd_endpoint(client, args):
_emit(client.get_endpoint_details(args.endpoint_id), args.json, _print_kv)
def cmd_sweep(client, args):
if args.company:
summaries = client.security_sweep(args.company)
else:
print("[INFO] No --company given; sweeping ALL client companies "
"(this makes many live API calls).", file=sys.stderr)
summaries = client.security_sweep_all_clients()
if args.json:
print(json.dumps([dataclasses.asdict(s) for s in summaries], indent=2))
else:
_print_sweep_table(summaries)
def _require_company_for_sweep() -> str:
from gz_client import ACG_COMPANIES_CONTAINER_ID
print("[INFO] No --company given; sweeping the ACG companies container.",
file=sys.stderr)
return ACG_COMPANIES_CONTAINER_ID
def cmd_policies(client, args):
_emit(client.list_policies(), args.json, _print_policy_table)
def cmd_policy(client, args):
print("[WARNING] Public API returns shallow policy detail only "
"(no granular config).", file=sys.stderr)
_emit(client.get_policy_details(args.policy_id), args.json, _print_kv)
def cmd_packages(client, args):
_emit(client.list_packages(), args.json, _print_package_table)
def cmd_quarantine(client, args):
_emit(client.list_quarantine(args.company), args.json, _print_quarantine_table)
def cmd_blocklist(client, args):
_emit(client.list_blocklist(args.company, page=args.page, per_page=args.per_page),
args.json, _print_blocklist_table)
def cmd_incidents(client, args):
_emit(client.list_incidents(args.company, page=args.page,
per_page=args.per_page),
args.json, _print_incidents_table)
def cmd_inventory(client, args):
_emit(client.get_inventory(refresh=args.refresh), args.json,
_print_inventory_table)
def cmd_create_package(client, args):
result = client.create_package(
package_name=args.name,
company_id=args.company,
description=args.description,
language=args.language,
)
_emit({"created": args.name, "result": result}, args.json, _print_kv)
def cmd_install_links(client, args):
_emit(client.get_installation_links(args.package, args.company),
args.json, _print_kv)
def cmd_scan(client, args):
result = client.create_scan_task(
target_ids=args.targets, scan_type=args.type, name=args.name
)
_emit({"scanTask": result}, args.json, _print_kv)
def cmd_move(client, args):
result = client.move_endpoints(args.endpoints, args.group)
_emit({"moved": args.endpoints, "to": args.group, "result": result},
args.json, _print_kv)
def cmd_make_group(client, args):
result = client.create_custom_group(args.name, args.parent)
_emit({"createdGroup": args.name, "result": result}, args.json, _print_kv)
# Substrings that mark a JSON-RPC method as state-destroying. `raw` can reach
# any method (incl. UNVERIFIED ones), so gate these behind --confirm too.
# isolate / blocklist add+remove are NEW destructive verbs from the incidents
# (EDR) module — gate them in `raw` as well as via the dedicated subcommands.
DESTRUCTIVE_RAW_PATTERNS = ("delete", "createuninstall", "createremove",
"createreconfigure", "isolat", "addtoblocklist",
"removefromblocklist")
def _is_destructive_method(method: str) -> bool:
m = method.lower()
return any(pat in m for pat in DESTRUCTIVE_RAW_PATTERNS)
def cmd_raw(client, args):
if _is_destructive_method(args.method) and not args.confirm:
print(f"[WARNING] '{args.method}' looks destructive; refusing without "
"--confirm.", file=sys.stderr)
return 3
try:
params = json.loads(args.params) if args.params else {}
except json.JSONDecodeError as exc:
print(f"[ERROR] --params is not valid JSON: {exc}", file=sys.stderr)
return 2
if not isinstance(params, dict):
print("[ERROR] --params must be a JSON object.", file=sys.stderr)
return 2
result = client._jsonrpc_request(args.module, args.method, params)
print(json.dumps(result, indent=2, default=_json_default))
return 0
# --- destructive (gated) ------------------------------------------------------
def _gated(action_desc: str, confirm: bool) -> bool:
if not confirm:
print("[WARNING] Refusing destructive action without --confirm.")
print(f"[INFO] Would: {action_desc}")
return False
return True
def cmd_delete_endpoint(client, args):
if not _gated(f"delete endpoint {args.endpoint_id}", args.confirm):
return 3
result = client.delete_endpoint(args.endpoint_id)
_emit({"deletedEndpoint": args.endpoint_id, "result": result},
args.json, _print_kv)
return 0
def cmd_delete_package(client, args):
if not _gated(f"delete package '{args.package}'", args.confirm):
return 3
result = client.delete_package(args.package, args.company)
_emit({"deletedPackage": args.package, "result": result}, args.json, _print_kv)
return 0
def cmd_delete_group(client, args):
if not _gated(f"delete custom group {args.group}", args.confirm):
return 3
result = client.delete_custom_group(args.group)
_emit({"deletedGroup": args.group, "result": result}, args.json, _print_kv)
return 0
# --- EDR / incident response (gated) ------------------------------------------
def cmd_isolate(client, args):
targets = ",".join(args.endpoints)
if not _gated(f"isolate endpoints {targets}", args.confirm):
return 3
result = client.isolate_endpoints(args.endpoints)
_emit({"isolated": args.endpoints, "result": result}, args.json, _print_kv)
return 0
def cmd_unisolate(client, args):
targets = ",".join(args.endpoints)
if not _gated(f"restore endpoints from isolation {targets}", args.confirm):
return 3
result = client.restore_endpoints_from_isolation(args.endpoints)
_emit({"unisolated": args.endpoints, "result": result}, args.json, _print_kv)
return 0
def cmd_blocklist_add(client, args):
desc = (f"add {len(args.hashes)} hash(es) to blocklist for company "
f"{args.company}: {','.join(args.hashes)}")
if not _gated(desc, args.confirm):
return 3
result = client.add_to_blocklist(
company_id=args.company,
hash_list=args.hashes,
hash_type=args.hash_type,
source_info=args.source_info,
)
_emit({"blocklistAdded": args.hashes, "company": args.company,
"result": result}, args.json, _print_kv)
return 0
def cmd_blocklist_remove(client, args):
if not _gated(f"remove blocklist item {args.id}", args.confirm):
return 3
result = client.remove_from_blocklist(args.id)
_emit({"blocklistRemoved": args.id, "result": result}, args.json, _print_kv)
return 0
# --- parser -------------------------------------------------------------------
def build_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(
prog="gz.py",
description="GravityZone Cloud Public API CLI (ACG MSP tenant).",
)
# --json is shared by every subcommand (via parents) so it is accepted
# after the subcommand, e.g. `gz.py companies --json`.
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--json", action="store_true", help="Emit raw JSON output.")
sub = p.add_subparsers(dest="command", required=True)
sub.add_parser("status", help="API key + license status.", parents=[common])
sub.add_parser("companies", help="List client companies.", parents=[common])
sp = sub.add_parser("endpoints", help="List endpoints under a company/group.",
parents=[common])
sp.add_argument("--company", help="Parent company/group id.")
sp.add_argument("--per-page", type=int, default=100)
sp = sub.add_parser("endpoint", help="Full detail for one endpoint.",
parents=[common])
sp.add_argument("endpoint_id")
sp = sub.add_parser("sweep", help="Live security posture sweep.", parents=[common])
sp.add_argument("--company", help="Parent id (defaults to ACG container).")
sub.add_parser("policies", help="List policies (id + name).", parents=[common])
sp = sub.add_parser("policy", help="Shallow detail for one policy.",
parents=[common])
sp.add_argument("policy_id")
sub.add_parser("packages", help="List installation packages.", parents=[common])
sp = sub.add_parser("quarantine", help="List quarantine items for a company.",
parents=[common])
sp.add_argument("--company", required=True)
sp = sub.add_parser("blocklist", help="List blocklisted hash items (EDR).",
parents=[common])
sp.add_argument("--company", help="Scope to one company id (optional).")
sp.add_argument("--page", type=int, default=1)
sp.add_argument("--per-page", type=int, default=100)
sp = sub.add_parser("incidents", help="List incidents under a company (EDR).",
parents=[common])
sp.add_argument("--company", required=True,
help="Parent company/group id (parentId; required).")
sp.add_argument("--page", type=int, default=1)
sp.add_argument("--per-page", type=int, default=500,
help="incidents requires 500-10000 (API constraint).")
sp = sub.add_parser("inventory", help="Show cached identity/structure.",
parents=[common])
sp.add_argument("--refresh", action="store_true", help="Force a full re-pull.")
sp = sub.add_parser("create-package", help="Create an installer package.",
parents=[common])
sp.add_argument("--name", required=True)
sp.add_argument("--company")
sp.add_argument("--description")
sp.add_argument("--language")
sp = sub.add_parser("install-links", help="Get installer download URLs.",
parents=[common])
sp.add_argument("--package", required=True)
sp.add_argument("--company")
sp = sub.add_parser("scan", help="Create a scan task.", parents=[common])
sp.add_argument("--targets", nargs="+", required=True)
sp.add_argument("--type", type=int, required=True,
help="1=Quick 2=Full 3=Memory 4=Custom (verify in console).")
sp.add_argument("--name")
sp = sub.add_parser("move", help="Move endpoints into a group.", parents=[common])
sp.add_argument("--endpoints", nargs="+", required=True)
sp.add_argument("--group", required=True)
sp = sub.add_parser("make-group", help="Create a custom group.", parents=[common])
sp.add_argument("--name", required=True)
sp.add_argument("--parent")
sp = sub.add_parser("raw", help="Call any method directly (power use).",
parents=[common])
sp.add_argument("--module", required=True)
sp.add_argument("--method", required=True)
sp.add_argument("--params", default="{}", help="JSON object of params.")
sp.add_argument("--confirm", action="store_true",
help="Required for destructive methods (delete/uninstall/"
"remove/reconfigure).")
# destructive (gated)
sp = sub.add_parser("delete-endpoint", help="Delete an endpoint (gated).",
parents=[common])
sp.add_argument("endpoint_id")
sp.add_argument("--confirm", action="store_true")
sp = sub.add_parser("delete-package", help="Delete a package (gated).",
parents=[common])
sp.add_argument("--package", required=True)
sp.add_argument("--company")
sp.add_argument("--confirm", action="store_true")
sp = sub.add_parser("delete-group", help="Delete a custom group (gated).",
parents=[common])
sp.add_argument("--group", required=True)
sp.add_argument("--confirm", action="store_true")
# EDR / incident response (gated)
sp = sub.add_parser("isolate",
help="Isolate endpoints from the network (gated).",
parents=[common])
sp.add_argument("--endpoints", nargs="+", required=True,
help="One or more endpoint ids (max 1000).")
sp.add_argument("--confirm", action="store_true")
sp = sub.add_parser("unisolate",
help="Restore endpoints from isolation (gated).",
parents=[common])
sp.add_argument("--endpoints", nargs="+", required=True,
help="One or more endpoint ids (max 1000).")
sp.add_argument("--confirm", action="store_true")
sp = sub.add_parser("blocklist-add",
help="Add hashes to the blocklist (gated).",
parents=[common])
sp.add_argument("--company", required=True)
sp.add_argument("--hashes", nargs="+", required=True,
help="One or more hash strings to block.")
sp.add_argument("--hash-type", type=int, default=1,
help="Hash type int (1 common; see console / API docs).")
sp.add_argument("--source-info", default="",
help="Free-text description of the source.")
sp.add_argument("--confirm", action="store_true")
sp = sub.add_parser("blocklist-remove",
help="Remove one blocklist entry (gated).",
parents=[common])
sp.add_argument("--id", required=True,
help="hashItemId — the 'id' from `blocklist` output.")
sp.add_argument("--confirm", action="store_true")
return p
HANDLERS = {
"status": cmd_status,
"companies": cmd_companies,
"endpoints": cmd_endpoints,
"endpoint": cmd_endpoint,
"sweep": cmd_sweep,
"policies": cmd_policies,
"policy": cmd_policy,
"packages": cmd_packages,
"quarantine": cmd_quarantine,
"blocklist": cmd_blocklist,
"incidents": cmd_incidents,
"inventory": cmd_inventory,
"create-package": cmd_create_package,
"install-links": cmd_install_links,
"scan": cmd_scan,
"move": cmd_move,
"make-group": cmd_make_group,
"raw": cmd_raw,
"delete-endpoint": cmd_delete_endpoint,
"delete-package": cmd_delete_package,
"delete-group": cmd_delete_group,
"isolate": cmd_isolate,
"unisolate": cmd_unisolate,
"blocklist-add": cmd_blocklist_add,
"blocklist-remove": cmd_blocklist_remove,
}
def main(argv=None) -> int:
args = build_parser().parse_args(argv)
handler = HANDLERS[args.command]
try:
client = GravityZoneClient()
rc = handler(client, args)
return rc if isinstance(rc, int) else 0
except GravityZoneError as exc:
print(f"[ERROR] {exc}", file=sys.stderr)
return 1
except KeyboardInterrupt:
return 130
if __name__ == "__main__":
raise SystemExit(main())