#!/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 python gz.py sweep --company python gz.py policies python gz.py policy python gz.py packages python gz.py quarantine --company python gz.py inventory --refresh python gz.py create-package --name "Win Default" --company python gz.py install-links --package "Win Default" --company python gz.py scan --targets --type 2 --name "Full scan" python gz.py move --endpoints --group python gz.py make-group --name "New Group" --parent python gz.py delete-endpoint --confirm python gz.py blocklist --company python gz.py incidents --company python gz.py isolate --endpoints --confirm python gz.py unisolate --endpoints --confirm python gz.py blocklist-add --company --hashes

--confirm python gz.py blocklist-remove --id --confirm python gz.py raw --module network --method getEndpointsList --params '{"page":1}' """ from __future__ import annotations import argparse import dataclasses import json import os import subprocess import sys from gz_client import GravityZoneClient, GravityZoneError, GZEndpointSummary 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 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) -> None: # Tolerant: some API methods return a list (e.g. installation links, # endpoint tags) rather than a dict. Render either cleanly. if isinstance(d, list): for i, item in enumerate(d): if isinstance(item, dict): if i: print(" ---") for k, v in item.items(): print(f" {k}: {v}") else: print(f" {item}") return if not isinstance(d, dict): print(f" {d}") return 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_reports_table(data: dict) -> None: items = data.get("items", []) print(f"Reports: {data.get('total', len(items))}") for r in items: print(f" {str(r.get('id','?')):26} {str(r.get('name','')):40} " f"type={r.get('type','')}") def _print_accounts_table(data: dict) -> None: items = data.get("items", []) print(f"Accounts: {data.get('total', len(items))}") for a in items: prof = a.get("profile", {}) or {} print(f" {str(a.get('id','?')):26} {str(a.get('email','')):34} " f"{prof.get('fullName','')}") def _print_scan_tasks_table(data: dict) -> None: items = data.get("items", []) print(f"Scan tasks: {data.get('total', len(items))}") for t in items: print(f" {str(t.get('id','?')):26} {str(t.get('name','')):30} " f"status={t.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_company(client, args): _emit(client.get_company_details(args.company_id), args.json, _print_kv) def cmd_company_by_user(client, args): _emit(client.get_company_by_user(args.username), args.json, _print_kv) def cmd_company_create(client, args): extra, rc = _load_json_arg(args.extra_json, "extra-json") if rc: return rc label = {0: "Partner", 1: "Customer"}.get(args.type, str(args.type)) if not _gated(f"create {label} company '{args.name}'", args.confirm): return 3 result = client.create_company(args.type, args.name, parent_id=args.parent, extra=extra or None) _emit({"createdCompany": args.name, "result": result}, args.json, _print_kv) return 0 def cmd_company_suspend(client, args): if not _gated(f"suspend company {args.id}", args.confirm): return 3 _emit({"suspended": args.id, "result": client.suspend_company(args.id)}, args.json, _print_kv) return 0 def cmd_company_activate(client, args): if not _gated(f"activate company {args.id}", args.confirm): return 3 _emit({"activated": args.id, "result": client.activate_company(args.id)}, args.json, _print_kv) return 0 def cmd_company_delete(client, args): if not _gated(f"delete company {args.id}", args.confirm): return 3 _emit({"deletedCompany": args.id, "result": client.delete_company(args.id)}, args.json, _print_kv) return 0 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): # getPolicyDetails returns the FULL granular module configuration (verified # live 2026-06-21). Use --json for the complete settings tree; the table # view shows the top-level keys only. _emit(client.get_policy_details(args.policy_id), args.json, _print_kv) def cmd_reports(client, args): _emit(client.list_reports(page=args.page, per_page=args.per_page), args.json, _print_reports_table) def cmd_accounts(client, args): _emit(client.list_accounts(page=args.page, per_page=args.per_page), args.json, _print_accounts_table) def cmd_notif_settings(client, args): _emit(client.get_notifications_settings(), args.json, _print_kv) def cmd_account(client, args): _emit(client.get_account_details(args.account_id), args.json, _print_kv) def _load_json_arg(raw, label): """Parse a JSON-object CLI arg; returns (obj, error_rc). error_rc is None on ok.""" if raw is None: return {}, None try: obj = json.loads(raw) except json.JSONDecodeError as exc: print(f"[ERROR] --{label} is not valid JSON: {exc}", file=sys.stderr) return None, 2 if not isinstance(obj, dict): print(f"[ERROR] --{label} must be a JSON object.", file=sys.stderr) return None, 2 return obj, None def cmd_account_create(client, args): rights, rc = _load_json_arg(args.rights_json, "rights-json") if rc: return rc extra, rc = _load_json_arg(args.extra_json, "extra-json") if rc: return rc profile = {} if args.full_name: profile["fullName"] = args.full_name if args.language: profile["language"] = args.language if args.timezone: profile["timezone"] = args.timezone if not _gated(f"create account {args.email} (role={args.role})", args.confirm): return 3 result = client.create_account( email=args.email, password=args.password, username=args.username, role=args.role, profile=profile or None, rights=rights or None, extra=extra or None, ) _emit({"createdAccount": args.email, "result": result}, args.json, _print_kv) return 0 def cmd_account_update(client, args): fields, rc = _load_json_arg(args.set_json, "set-json") if rc: return rc if not fields: print("[ERROR] --set-json (object of fields to change) is required.", file=sys.stderr) return 2 if not _gated(f"update account {args.id} fields={list(fields)}", args.confirm): return 3 result = client.update_account(args.id, fields) _emit({"updatedAccount": args.id, "result": result}, args.json, _print_kv) return 0 def cmd_account_delete(client, args): if not _gated(f"delete account {args.id}", args.confirm): return 3 result = client.delete_account(args.id) _emit({"deletedAccount": args.id, "result": result}, args.json, _print_kv) return 0 def cmd_notif_configure(client, args): settings, rc = _load_json_arg(args.settings_json, "settings-json") if rc: return rc if not settings: print("[ERROR] --settings-json (the settings object) is required.", file=sys.stderr) return 2 if not _gated(f"configure notification settings ({list(settings)})", args.confirm): return 3 result = client.configure_notifications_settings(settings) _emit({"notificationsConfigured": True, "result": result}, args.json, _print_kv) return 0 def cmd_scan_tasks(client, args): _emit(client.list_scan_tasks(page=args.page, per_page=args.per_page), args.json, _print_scan_tasks_table) def cmd_assign_policy(client, args): desc = (f"assign policy {args.policy} to {len(args.targets)} target(s): " f"{','.join(args.targets)}") if not _gated(desc, args.confirm): return 3 result = client.assign_policy( args.policy, args.targets, force_inheritance=args.force_inheritance, ) _emit({"assignedPolicy": args.policy, "targets": args.targets, "result": result}, args.json, _print_kv) return 0 def _push_read(emit_fn) -> int: """Run a push read, treating 'never configured' as an expected (non-error) state rather than a failure (so it does not pollute errorlog).""" try: emit_fn() return 0 except GravityZoneError as exc: msg = str(exc).lower() if "not set" in msg or "are not" in msg or "not available" in msg: print("[INFO] Push event service is not configured on this tenant.") return 0 raise def cmd_push_settings(client, args): return _push_read( lambda: _emit(client.get_push_settings(), args.json, _print_kv) ) def cmd_push_stats(client, args): return _push_read( lambda: _emit(client.get_push_stats(), args.json, _print_kv) ) def cmd_push_set(client, args): state = "ENABLE" if args.status == 1 else "DISABLE" if args.status == 1 and not args.url: print("[ERROR] --url is required to enable the push event service.", file=sys.stderr) return 2 desc = f"{state} GravityZone push event service (url={args.url or '-'})" if not _gated(desc, args.confirm): return 3 result = client.set_push_settings( status=args.status, service_type=args.service_type, url=args.url, require_valid_ssl=not args.allow_insecure_ssl, authorization=args.authorization, ) _emit({"pushService": state, "result": result}, args.json, _print_kv) return 0 def cmd_push_test(client, args): if not _gated(f"send test push event '{args.event_type}'", args.confirm): return 3 extra, rc = _load_json_arg(args.extra_json, "extra-json") if rc: return rc result = client.send_test_push_event(args.event_type, extra=extra or None) _emit({"testEvent": args.event_type, "result": result}, args.json, _print_kv) return 0 def cmd_package_details(client, args): _emit(client.get_package_details(args.package_id), args.json, _print_kv) def cmd_report_create(client, args): extra, rc = _load_json_arg(args.extra_json, "extra-json") if rc: return rc if not _gated(f"create report '{args.name}'", args.confirm): return 3 result = client.create_report(args.name, extra=extra or None) _emit({"createdReport": args.name, "result": result}, args.json, _print_kv) return 0 def cmd_report_links(client, args): _emit(client.get_report_links(args.id), args.json, _print_kv) def cmd_report_delete(client, args): if not _gated(f"delete report {args.id}", args.confirm): return 3 _emit({"deletedReport": args.id, "result": client.delete_report(args.id)}, args.json, _print_kv) return 0 def cmd_quarantine_remove(client, args): if not _gated(f"remove {len(args.items)} quarantine item(s)", args.confirm): return 3 result = client.remove_quarantine_items(args.items) _emit({"removedQuarantine": args.items, "result": result}, args.json, _print_kv) return 0 def cmd_quarantine_restore(client, args): extra, rc = _load_json_arg(args.extra_json, "extra-json") if rc: return rc if not _gated(f"restore {len(args.items)} quarantine item(s)", args.confirm): return 3 result = client.restore_quarantine_items(args.items, extra=extra or None) _emit({"restoredQuarantine": args.items, "result": result}, args.json, _print_kv) return 0 def cmd_custom_rules(client, args): _emit(client.list_custom_rules(page=args.page, per_page=args.per_page), args.json, _print_kv) def cmd_custom_rule_create(client, args): extra, rc = _load_json_arg(args.extra_json, "extra-json") if rc: return rc if not _gated(f"create custom rule '{args.name}'", args.confirm): return 3 result = client.create_custom_rule(args.name, extra=extra or None) _emit({"createdRule": args.name, "result": result}, args.json, _print_kv) return 0 def cmd_custom_rule_delete(client, args): if not _gated(f"delete custom rule {args.id}", args.confirm): return 3 _emit({"deletedRule": args.id, "result": client.delete_custom_rule(args.id)}, args.json, _print_kv) return 0 def cmd_incident_status(client, args): fields, rc = _load_json_arg(args.set_json, "set-json") if rc: return rc if not _gated(f"change incident status (type={args.type})", args.confirm): return 3 result = client.change_incident_status(args.type, fields) _emit({"incidentStatus": "changed", "result": result}, args.json, _print_kv) return 0 def cmd_incident_note(client, args): fields, rc = _load_json_arg(args.set_json, "set-json") if rc: return rc if not _gated(f"update incident note (type={args.type})", args.confirm): return 3 result = client.update_incident_note(args.type, fields) _emit({"incidentNote": "updated", "result": result}, args.json, _print_kv) return 0 def cmd_monthly_usage(client, args): _emit(client.get_monthly_usage(), args.json, _print_kv) def cmd_integrations(client, args): _emit(client.get_configured_integrations(page=args.page, per_page=args.per_page), 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 _print_tags(tags) -> None: items = tags if isinstance(tags, list) else (tags or {}).get("items", []) print(f"Endpoint tags: {len(items)}") for t in items: print(f" {t}") def cmd_endpoint_tags(client, args): _emit(client.get_endpoint_tags(), args.json, _print_tags) def cmd_set_label(client, args): if not _gated(f"label endpoint {args.endpoint} = '{args.label}'", args.confirm): return 3 result = client.set_endpoint_label(args.endpoint, args.label) _emit({"labeled": args.endpoint, "label": args.label, "result": result}, args.json, _print_kv) return 0 def cmd_reconfigure(client, args): extra, rc = _load_json_arg(args.extra_json, "extra-json") if rc: return rc if not _gated(f"reconfigure {len(args.targets)} agent(s): {','.join(args.targets)}", args.confirm): return 3 result = client.reconfigure_client(args.targets, extra=extra or None) _emit({"reconfigured": args.targets, "result": result}, args.json, _print_kv) return 0 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", "assignpolicy", "setpushevent", "createaccount", "updateaccount", "configurenotif", "createcompany", "suspendcompany", "activatecompany", "setendpointlabel", "createreport", "createrestore", "createcustomrule", "changeincident", "updateincident", "sendtestpush") 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.id}", args.confirm): return 3 result = client.delete_package(args.id) _emit({"deletedPackage": args.id, "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("company", help="Company detail (no id = own company).", parents=[common]) sp.add_argument("company_id", nargs="?", help="Company id (optional).") sp = sub.add_parser("company-by-user", help="Company that owns a username.", parents=[common]) sp.add_argument("--username", required=True) sp = sub.add_parser("company-create", help="Create a company (gated).", parents=[common]) sp.add_argument("--type", type=int, required=True, help="0=Partner, 1=Customer.") sp.add_argument("--name", required=True) sp.add_argument("--parent", help="parentId (defaults to the key's company).") sp.add_argument("--extra-json", help="JSON object of extra fields (licenseSubscription, " "address, assignedProductType, ...).") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("company-suspend", help="Suspend a company (gated).", parents=[common]) sp.add_argument("--id", required=True) sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("company-activate", help="Activate a company (gated).", parents=[common]) sp.add_argument("--id", required=True) sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("company-delete", help="Delete a company (gated).", parents=[common]) sp.add_argument("--id", required=True) sp.add_argument("--confirm", action="store_true") 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="Full granular config for one policy (use --json).", parents=[common]) sp.add_argument("policy_id") sub.add_parser("packages", help="List installation packages.", parents=[common]) sp = sub.add_parser("reports", help="List saved reports.", parents=[common]) sp.add_argument("--page", type=int, default=1) sp.add_argument("--per-page", type=int, default=100) sp = sub.add_parser("accounts", help="List GravityZone console accounts.", parents=[common]) sp.add_argument("--page", type=int, default=1) sp.add_argument("--per-page", type=int, default=100) sub.add_parser("notif-settings", help="Show notification settings.", parents=[common]) sp = sub.add_parser("account", help="Account detail (no id = the API key's own account).", parents=[common]) sp.add_argument("account_id", nargs="?", help="Account id (optional).") sp = sub.add_parser("scan-tasks", help="List scan tasks.", parents=[common]) sp.add_argument("--page", type=int, default=1) sp.add_argument("--per-page", type=int, default=100) sub.add_parser("push-settings", help="Show push event service settings.", parents=[common]) sub.add_parser("push-stats", help="Show push event service delivery stats.", parents=[common]) sp = sub.add_parser("package-details", help="Installation package detail.", parents=[common]) sp.add_argument("package_id") sub.add_parser("monthly-usage", help="Monthly license usage.", parents=[common]) sp = sub.add_parser("integrations", help="List configured integrations.", parents=[common]) sp.add_argument("--page", type=int, default=1) sp.add_argument("--per-page", type=int, default=100) sp = sub.add_parser("custom-rules", help="List EDR custom rules.", parents=[common]) sp.add_argument("--page", type=int, default=1) sp.add_argument("--per-page", type=int, default=100) sp = sub.add_parser("report-links", help="Get a report's download links.", parents=[common]) sp.add_argument("--id", required=True, help="reportId.") 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") sub.add_parser("endpoint-tags", help="List endpoint tags.", parents=[common]) sp = sub.add_parser("set-label", help="Set an endpoint's label (gated).", parents=[common]) sp.add_argument("--endpoint", required=True, help="endpointId.") sp.add_argument("--label", required=True) sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("reconfigure", help="Reconfigure installed agents (gated).", parents=[common]) sp.add_argument("--targets", nargs="+", required=True, help="Endpoint ids.") sp.add_argument("--extra-json", help="JSON reconfigure body (modules/roles/scanMode).") sp.add_argument("--confirm", action="store_true") 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 by id (gated).", parents=[common]) sp.add_argument("--id", required=True, help="packageId (from `packages`).") 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") sp = sub.add_parser("assign-policy", help="Assign an existing policy to endpoints/groups (gated).", parents=[common]) sp.add_argument("--policy", required=True, help="policyId to assign.") sp.add_argument("--targets", nargs="+", required=True, help="One or more endpoint/group ids.") sp.add_argument("--force-inheritance", action="store_true", help="Force policy inheritance to sub-items.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("account-create", help="Create a console account (gated).", parents=[common]) sp.add_argument("--email", required=True) sp.add_argument("--password") sp.add_argument("--username") sp.add_argument("--role", type=int, help="Numeric role id (see docs/console).") sp.add_argument("--full-name") sp.add_argument("--language", help="e.g. en_US") sp.add_argument("--timezone", help="e.g. America/Phoenix") sp.add_argument("--rights-json", help="JSON object of rights flags.") sp.add_argument("--extra-json", help="JSON object of any extra documented fields.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("account-update", help="Update a console account (gated).", parents=[common]) sp.add_argument("--id", required=True, help="accountId.") sp.add_argument("--set-json", required=True, help="JSON object of fields to change.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("account-delete", help="Delete a console account (gated).", parents=[common]) sp.add_argument("--id", required=True, help="accountId.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("notif-configure", help="Set notification settings (gated).", parents=[common]) sp.add_argument("--settings-json", required=True, help="JSON object of the notification settings to apply.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("push-test", help="Send a test push event (gated).", parents=[common]) sp.add_argument("--event-type", required=True) sp.add_argument("--extra-json") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("report-create", help="Create a report (gated).", parents=[common]) sp.add_argument("--name", required=True) sp.add_argument("--extra-json", help="JSON: type, targetIds, recurrence, format...") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("report-delete", help="Delete a report (gated).", parents=[common]) sp.add_argument("--id", required=True, help="reportId.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("quarantine-remove", help="Delete quarantined items (gated).", parents=[common]) sp.add_argument("--items", nargs="+", required=True, help="quarantineItemsIds.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("quarantine-restore", help="Restore quarantined items (gated).", parents=[common]) sp.add_argument("--items", nargs="+", required=True, help="quarantineItemsIds.") sp.add_argument("--extra-json", help="JSON: addExclusionInPolicy, etc.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("custom-rule-create", help="Create an EDR custom rule (gated).", parents=[common]) sp.add_argument("--name", required=True) sp.add_argument("--extra-json", help="JSON: settings, companyId, tags...") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("custom-rule-delete", help="Delete an EDR custom rule (gated).", parents=[common]) sp.add_argument("--id", required=True, help="ruleId.") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("incident-status", help="Change an incident's status (gated).", parents=[common]) sp.add_argument("--type", required=True, help="Incident type/category.") sp.add_argument("--set-json", required=True, help="JSON: id, status, ...") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("incident-note", help="Update an incident note (gated).", parents=[common]) sp.add_argument("--type", required=True, help="Incident type/category.") sp.add_argument("--set-json", required=True, help="JSON: id, note, ...") sp.add_argument("--confirm", action="store_true") sp = sub.add_parser("push-set", help="Configure the push event service (gated).", parents=[common]) sp.add_argument("--status", type=int, required=True, choices=[0, 1], help="1=enable, 0=disable.") sp.add_argument("--url", help="Receiver URL GravityZone POSTs events to " "(required to enable).") sp.add_argument("--service-type", default="jsonRPC", help="jsonRPC|splunk|cef (default jsonRPC).") sp.add_argument("--authorization", help="Optional Authorization header the receiver expects.") sp.add_argument("--allow-insecure-ssl", action="store_true", help="Do not require a valid SSL cert on the receiver.") sp.add_argument("--confirm", action="store_true") return p HANDLERS = { "status": cmd_status, "companies": cmd_companies, "company": cmd_company, "company-by-user": cmd_company_by_user, "company-create": cmd_company_create, "company-suspend": cmd_company_suspend, "company-activate": cmd_company_activate, "company-delete": cmd_company_delete, "endpoints": cmd_endpoints, "endpoint": cmd_endpoint, "sweep": cmd_sweep, "policies": cmd_policies, "policy": cmd_policy, "packages": cmd_packages, "reports": cmd_reports, "accounts": cmd_accounts, "notif-settings": cmd_notif_settings, "account": cmd_account, "account-create": cmd_account_create, "account-update": cmd_account_update, "account-delete": cmd_account_delete, "notif-configure": cmd_notif_configure, "scan-tasks": cmd_scan_tasks, "push-settings": cmd_push_settings, "push-stats": cmd_push_stats, "assign-policy": cmd_assign_policy, "push-set": cmd_push_set, "push-test": cmd_push_test, "package-details": cmd_package_details, "monthly-usage": cmd_monthly_usage, "integrations": cmd_integrations, "custom-rules": cmd_custom_rules, "custom-rule-create": cmd_custom_rule_create, "custom-rule-delete": cmd_custom_rule_delete, "incident-status": cmd_incident_status, "incident-note": cmd_incident_note, "report-create": cmd_report_create, "report-links": cmd_report_links, "report-delete": cmd_report_delete, "quarantine-remove": cmd_quarantine_remove, "quarantine-restore": cmd_quarantine_restore, "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, "endpoint-tags": cmd_endpoint_tags, "set-label": cmd_set_label, "reconfigure": cmd_reconfigure, "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) _log_skill_error("bitdefender", f"{exc}", context=f"cmd={getattr(args, 'command', '?')}") return 1 except KeyboardInterrupt: return 130 if __name__ == "__main__": raise SystemExit(main())