f3nris
1d475a6cf5
Declared as a boolean, isActive rendered as a checkbox in the toolbox, and a checkbox has no empty state: the count was forced to one half of the fleet or the other, with no way to ask for both. As text, an empty field means the filter is not sent. true/yes/1 and false/no/0 are read in any case, a real JSON boolean from a playbook still works, and anything else is refused rather than folded into false — that would answer a different question than the one asked.
69 lines
2.9 KiB
Python
69 lines
2.9 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", "{}"))
|
|
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"))),
|
|
# Agents strictly newer than the given version, e.g. 23.4.2.6 — the console
|
|
# answers with the fleet still trailing behind a target build.
|
|
"agentVersion__gt": inputs.get("agentVersion__gt"),
|
|
# 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)
|