f3nris
505c518bf8
The agent collection is DRF-paginated, so `count` already carries the filtered total. endpoint_count asks for limit=1 and reads it, instead of pulling up to 10000 agent records back through the playbook the way endpoint_search does — and endpoint_search only filters on hostname. Filters cover the documented agent query parameters (hostname, network, os, version, status, isolation_state, tenant), add lastseen/firstseen bounds over the DRF range lookups, and keep a raw `filters` escape hatch for anything else the instance accepts. The API silently drops a query parameter it does not recognise as a filter, which turns a typo into a plausible-looking fleet-wide total, so the fleet `total` is returned next to `count` to make that case visible. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
113 lines
3.6 KiB
Python
113 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=60) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else {}
|
|
|
|
|
|
def qs(params):
|
|
clean = {}
|
|
for k, v in params.items():
|
|
if v is None or v == "" or v == []:
|
|
continue
|
|
if isinstance(v, bool):
|
|
v = "true" if v else "false"
|
|
clean[k] = v
|
|
return ("?" + urllib.parse.urlencode(clean)) if clean else ""
|
|
|
|
|
|
def csv(v):
|
|
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
|
|
|
|
|
# Agent fields HarfangLab exposes as exact-match query filters on
|
|
# /api/data/endpoint/Agent/. Input name == API filter name.
|
|
EXACT_FILTERS = [
|
|
"hostname",
|
|
"domainname",
|
|
"domain",
|
|
"ipaddress",
|
|
"external_ipaddress",
|
|
"ostype",
|
|
"osproducttype",
|
|
"osversion",
|
|
"producttype",
|
|
"version",
|
|
"status",
|
|
"isolation_state",
|
|
"tenant",
|
|
]
|
|
|
|
# Convenience bounds mapped onto Django REST Framework range lookups.
|
|
RANGE_FILTERS = {
|
|
"lastseen_after": "lastseen__gte",
|
|
"lastseen_before": "lastseen__lte",
|
|
"firstseen_after": "firstseen__gte",
|
|
"firstseen_before": "firstseen__lte",
|
|
}
|
|
|
|
|
|
def to_bool(v):
|
|
if isinstance(v, bool):
|
|
return v
|
|
return str(v).strip().lower() in ("true", "1", "yes")
|
|
|
|
|
|
def main():
|
|
secrets = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
base = secrets.get("url", "").rstrip("/")
|
|
headers = {
|
|
"Authorization": "Token " + secrets.get("api_token", ""),
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
}
|
|
# === per-command logic ===
|
|
filters = {}
|
|
for field in EXACT_FILTERS:
|
|
value = inputs.get(field)
|
|
if value is None or value == "":
|
|
continue
|
|
filters[field] = to_bool(value) if field == "isolation_state" else value
|
|
for field, lookup in RANGE_FILTERS.items():
|
|
value = inputs.get(field)
|
|
if value:
|
|
filters[lookup] = value
|
|
# Escape hatch for any filter this command does not name explicitly,
|
|
# e.g. "groups__name=Serveurs&osmajor=10".
|
|
for tok in str(inputs.get("filters") or "").split("&"):
|
|
kv = tok.split("=", 1)
|
|
if len(kv) == 2 and kv[0].strip():
|
|
filters[kv[0].strip()] = kv[1]
|
|
|
|
# The agent list is a DRF paginated collection: `count` is the total number
|
|
# of matches, independent of the page size. limit=1 keeps the payload tiny.
|
|
agents = base + "/api/data/endpoint/Agent/"
|
|
matched = request("GET", agents + qs(dict(filters, limit=1, offset=0)), headers)
|
|
result = {"count": matched.get("count", 0), "filters": filters}
|
|
|
|
# The API silently drops a query parameter it does not recognise as a
|
|
# filter, which yields the unfiltered total dressed up as a filtered count.
|
|
# Returning the fleet total alongside makes that failure visible.
|
|
if filters:
|
|
everything = request("GET", agents + qs({"limit": 1, "offset": 0}), headers)
|
|
result["total"] = everything.get("count", 0)
|
|
else:
|
|
result["total"] = result["count"]
|
|
|
|
print(json.dumps(result))
|
|
|
|
|
|
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)
|