Files
riposte-marketplace/integrations/sentinelone/scripts/count_agents.py
T
f3nris c67343b7cf feat(sentinelone): instance-level account scope (1.20.0)
An instance can carry the account IDs of the client it serves, for clients
sharing one multi-tenant console. count_agents, get_threats, get_alerts and
get_unified_alerts use them when the block leaves its account filter empty;
an account named on the block still wins.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TgjV26rz6VcP262dX4vBrD
2026-10-19 19:08:10 +00:00

79 lines
3.6 KiB
Python

import json, os, sys, urllib.request, urllib.parse, urllib.error
def request(method, url, headers, body=None):
data = json.dumps(body).encode("utf-8") if body is not None else None
req = urllib.request.Request(url, data=data, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=30) as resp:
raw = resp.read()
return json.loads(raw) if raw else {}
def csv(v):
return [x.strip() for x in str(v or "").split(",") if x.strip()]
TRUE_WORDS = {"true", "yes", "1"}
FALSE_WORDS = {"false", "no", "0"}
def main():
secrets = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
# An instance bound to one account of a shared, multi-tenant console carries it
# in its config: it scopes the call unless the block names accounts itself.
if not str(inputs.get("accountIds") or "").strip() and str(secrets.get("account_ids") or "").strip():
inputs["accountIds"] = str(secrets["account_ids"])
base = secrets.get("url", "").rstrip("/") + "/web/api/v2.1"
headers = {
"Authorization": "ApiToken " + secrets.get("api_token", ""),
"Accept": "application/json",
"Content-Type": "application/json",
}
# === REQUEST ===
qs = {
"computerName": inputs.get("computer_name"),
"osTypes": inputs.get("os_type"),
"scan_status": inputs.get("scan_status"),
"siteIds": inputs.get("siteIds"),
"groupIds": inputs.get("groupIds"),
# Scopes the count to one or more accounts of a multi-tenant console.
"accountIds": ",".join(csv(inputs.get("accountIds"))),
# The four version comparators the console accepts, e.g. 23.4.2.6. Sent
# together they bound a range: __gte with __lte counts the fleet on a
# band of builds, __gt alone what is already past a target, __lte alone
# what is still to be upgraded to it. The strict and inclusive pairs
# differ only on the boundary version itself.
"agentVersion__gt": inputs.get("agentVersion__gt"),
"agentVersion__gte": inputs.get("agentVersion__gte"),
"agentVersion__lt": inputs.get("agentVersion__lt"),
"agentVersion__lte": inputs.get("agentVersion__lte"),
# Comma-separated lists are re-joined so a hand-typed "connected, disconnected"
# does not reach the API with the space inside the value.
"networkStatuses": ",".join(csv(inputs.get("networkStatuses"))),
"operationalStatesNin": ",".join(csv(inputs.get("operationalStatesNin"))),
"machineTypes": ",".join(csv(inputs.get("machineTypes"))),
}
# Text rather than a checkbox, which has no empty state: left blank the filter
# is not sent at all and the count covers active and inactive agents alike.
# A value we cannot read is refused rather than folded into false, which would
# answer a different question than the one asked.
active = str(inputs.get("isActive", "")).strip().lower()
if active:
if active not in TRUE_WORDS | FALSE_WORDS:
raise ValueError("isActive must be true or false, got: " + repr(inputs["isActive"]))
qs["isActive"] = "true" if active in TRUE_WORDS else "false"
url = base + "/agents/count?" + urllib.parse.urlencode({k: v for k, v in qs.items() if v not in (None, "")})
print(json.dumps(request("GET", url, headers)))
# === END ===
try:
main()
except urllib.error.HTTPError as e:
print(json.dumps({"error": "HTTP " + str(e.code), "detail": e.read().decode("utf-8", "replace")}))
sys.exit(1)
except Exception as e:
print(json.dumps({"error": str(e)}))
sys.exit(1)