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

sync: auto-sync from HOWARD-HOME at 2026-06-25 13:43:47

Browse Source 
Author: Howard Enos
Machine: HOWARD-HOME
Timestamp: 2026-06-25 13:43:47
This commit is contained in:
Howard Enos
2026-06-25 13:44:18 -07:00
parent 1852f755ad
commit a96a15a1d2
7 changed files with 333 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

100
.claude/skills/bitdefender/scripts/gz_client.py
View File

@@ -105,6 +105,10 @@ GRAVITYZONE_API_BASE_URL = os.environ.get(
GRAVITYZONE_TIMEOUT_SECONDS = 60.0
GRAVITYZONE_CONNECT_TIMEOUT_SECONDS = 10.0
# Hard ceiling on paginated loops: a misbehaving API that always returns a full
# page must never spin forever (the sweep also fans out N detail calls per page).
MAX_PAGINATION_PAGES = 1000
ACG_ROOT_COMPANY_ID = "5c4280716c0318f3478b456a"
ACG_COMPANIES_CONTAINER_ID = "5c4280716c0318f3478b456e"
@@ -257,7 +261,12 @@ class GravityZoneClient:
"""Make one JSON-RPC call. Returns body['result'] or raises GravityZoneError."""
url = f"{self.api_base_url}/{module}"
payload = {"id": "1", "jsonrpc": "2.0", "method": method, "params": params}
body = self._post(url, payload)
# Read methods (get*/list*) are safe to retry on timeout/5xx; state-changing
# methods are NOT (a timeout can fire after the server committed -> a retry
# would double-execute, e.g. createScanTask/createPackage). 429 is a
# pre-processing rate-limit reject and is always safe (handled in _post).
idempotent = method.lower().startswith(("get", "list"))
body = self._post(url, payload, idempotent=idempotent)
if isinstance(body, dict) and "error" in body and body["error"] is not None:
err = body["error"]
@@ -271,17 +280,21 @@ class GravityZoneClient:
return body.get("result")
return body
def _post(self, url: str, payload: dict) -> Any:
"""POST with bounded retry on transient failures (429/5xx/timeout)."""
def _post(self, url: str, payload: dict, idempotent: bool = False) -> Any:
"""POST with bounded retry on transient failures. A 429 (pre-processing
rate-limit reject, no side effect) is always retried; a timeout/5xx is
retried ONLY for idempotent (read) calls, so a non-idempotent write is
never silently re-executed after a server-side commit + timeout."""
data = json.dumps(payload).encode("utf-8")
for attempt in range(RETRY_MAX_ATTEMPTS):
try:
return self._post_once(url, data)
except _RetryableHTTP as exc:
if attempt >= RETRY_MAX_ATTEMPTS - 1:
retry_safe = (exc.code == 429) or idempotent
if not retry_safe or attempt >= RETRY_MAX_ATTEMPTS - 1:
note = "" if retry_safe else " (non-idempotent; not retried)"
raise GravityZoneError(
f"GravityZone HTTP {exc.code} after {RETRY_MAX_ATTEMPTS} "
f"attempts: {exc.detail}".rstrip(": ")
f"GravityZone HTTP {exc.code}{note}: {exc.detail}".rstrip(": ")
) from exc
delay = _retry_delay(exc.headers, attempt)
print(
@@ -302,7 +315,6 @@ class GravityZoneClient:
url, content=data, auth=(self.api_key, ""),
headers={"Content-Type": "application/json"})
resp.raise_for_status()
return resp.json()
except httpx.TimeoutException as exc:
raise _RetryableHTTP("timeout", detail=str(exc)) from exc
except httpx.HTTPStatusError as exc:
@@ -315,6 +327,12 @@ class GravityZoneClient:
except httpx.HTTPError as exc:
raise GravityZoneError(
f"GravityZone request failed: {exc}") from exc
try:
return resp.json()
except ValueError as exc:
body = (resp.text or "")[:ERROR_BODY_MAX_CHARS]
raise GravityZoneError(
f"GravityZone returned a non-JSON response: {body}") from exc
# stdlib fallback
token = base64.b64encode(f"{self.api_key}:".encode("utf-8")).decode("ascii")
@@ -330,7 +348,6 @@ class GravityZoneClient:
try:
with urllib.request.urlopen(req, timeout=self.timeout) as resp:
raw = resp.read()
return json.loads(raw.decode("utf-8"))
except urllib.error.HTTPError as exc:
detail = exc.read().decode("utf-8", errors="replace")[:ERROR_BODY_MAX_CHARS]
if exc.code in RETRY_STATUSES:
@@ -341,6 +358,12 @@ class GravityZoneClient:
raise _RetryableHTTP("timeout", detail=str(exc)) from exc
except urllib.error.URLError as exc:
raise GravityZoneError(f"GravityZone request failed: {exc}") from exc
try:
return json.loads(raw.decode("utf-8"))
except ValueError as exc:
body = raw.decode("utf-8", errors="replace")[:ERROR_BODY_MAX_CHARS]
raise GravityZoneError(
f"GravityZone returned a non-JSON response: {body}") from exc
# ======================================================================
# READ METHODS (always live)
@@ -511,6 +534,10 @@ class GravityZoneClient:
if len(items) < per_page:
break
page += 1
if page > MAX_PAGINATION_PAGES:
print(f"[WARNING] security_sweep hit the {MAX_PAGINATION_PAGES}-page "
"ceiling; results may be truncated.", file=sys.stderr)
break
summaries.sort(
key=lambda s: (
@@ -1252,6 +1279,11 @@ class GravityZoneClient:
if len(items) < 100:
break
page += 1
if page > MAX_PAGINATION_PAGES:
print("[WARNING] refresh_inventory hit the "
f"{MAX_PAGINATION_PAGES}-page ceiling for company "
f"{cid}; inventory may be truncated.", file=sys.stderr)
break
try:
for p in self.list_policies(per_page=100).get("items", []):
@@ -1287,29 +1319,39 @@ class GravityZoneClient:
return self.refresh_inventory()
def _cache_add_group(self, group_id: str, name: str) -> None:
with self._cache_lock():
cache = self._read_cache()
if cache is None:
return # no cache yet - next refresh picks it up
# Groups live in the inventory tree; store under a 'groups' map.
cache.setdefault("groups", {})[group_id] = name
self._write_cache(cache)
# Best-effort: the cache is only a hint, so a write failure must NEVER
# turn a successful API mutation (createCustomGroup) into a reported error.
try:
with self._cache_lock():
cache = self._read_cache()
if cache is None:
return # no cache yet - next refresh picks it up
# Groups live in the inventory tree; store under a 'groups' map.
cache.setdefault("groups", {})[group_id] = name
self._write_cache(cache)
except Exception:
pass
def _cache_add_package(self, package_name: str, create_result: Any) -> None:
with self._cache_lock():
cache = self._read_cache()
if cache is None:
return
packages = cache.setdefault("packages", [])
pkg_id = create_result if isinstance(create_result, str) else None
if isinstance(create_result, dict):
pkg_id = create_result.get("id")
if not any(
(isinstance(p, dict) and p.get("name") == package_name)
for p in packages
):
packages.append({"id": pkg_id, "name": package_name})
self._write_cache(cache)
# Best-effort (see _cache_add_group): never let a cache failure mask a
# successful createPackage.
try:
with self._cache_lock():
cache = self._read_cache()
if cache is None:
return
packages = cache.setdefault("packages", [])
pkg_id = create_result if isinstance(create_result, str) else None
if isinstance(create_result, dict):
pkg_id = create_result.get("id")
if not any(
(isinstance(p, dict) and p.get("name") == package_name)
for p in packages
):
packages.append({"id": pkg_id, "name": package_name})
self._write_cache(cache)
except Exception:
pass
def main() -> int: