feat(cortex-xdr): count endpoints by filter without listing them (1.7.0)

The endpoints grid already answers with reply.total_count — the number
of endpoints matching the filters, independent of the page asked for —
so endpoint_count asks for a one-record page and reads it, where
get-endpoints has to walk the estate 100 records at a time to arrive at
the same figure, and only filters on four fields.

Filters cover the twelve list fields the API accepts (hostname, alias,
endpoint id, internal and public IP, group, platform, distribution,
status, isolation, scan status, username), add first_seen/last_seen
bounds taking ISO8601 or epoch ms, and keep a raw JSON `filters` escape
hatch for anything else. Filters are ANDed; the API has no OR.

The estate `total` comes back next to `count` so the figure reads as a
proportion, and a filter too broad to mean anything shows as matching
everything rather than passing for an answer.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

integrations/cortex-xdr/scripts/endpoint_count.py

@@ -0,0 +1,156 @@
+import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error
+from datetime import datetime
+
+
+def _client():
+    s = json.loads(os.environ.get("INTEGRATION_SECRETS") or "{}")
+    raw = str(s.get("url") or "").strip().rstrip("/")
+    if not raw:
+        raise ValueError("no url configured — paste the tenant API URL (Cortex XDR > Settings > Configurations > API Keys > Copy URL)")
+    if "://" not in raw:
+        raw = "https://" + raw
+    # The tenant URL is a bare host. Drop whatever was pasted after it (a stray
+    # /public_api/v1, a console path) so the API root is built exactly once.
+    scheme, _, rest = raw.partition("://")
+    base = scheme + "://" + rest.split("/")[0] + "/public_api/v1"
+    key = s.get("api_key", "")
+    kid = str(s.get("api_key_id", ""))
+    if not key or not kid:
+        raise ValueError("api_key and api_key_id are both required")
+    # Nonce and timestamp ride along in both modes, as the reference client does.
+    # A standard key travels as-is; an advanced one as sha256(key + nonce + ts).
+    nonce = "".join(secrets.choice(string.ascii_letters + string.digits) for _ in range(64))
+    ts = str(int(time.time()) * 1000)
+    headers = {
+        "x-xdr-auth-id": kid,
+        "x-xdr-nonce": nonce,
+        "x-xdr-timestamp": ts,
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    if str(s.get("auth_type") or "standard").lower() == "advanced":
+        headers["Authorization"] = hashlib.sha256((key + nonce + ts).encode("utf-8")).hexdigest()
+    else:
+        headers["Authorization"] = key
+    return base, headers
+
+
+def _not_json(r, raw):
+    """A 2xx that is not JSON means we are not talking to the XDR API at all."""
+    ctype = (r.headers.get("Content-Type") or "unknown").split(";")[0].strip()
+    head = raw[:160].decode("utf-8", "replace").replace("\n", " ").strip()
+    return (
+        "expected JSON from " + r.geturl() + ", got " + ctype + " (HTTP " + str(r.status) + "): " + head
+        + " — check the configured url is the tenant API host"
+        + " (https://api-<tenant>.xdr.<region>.paloaltonetworks.com), not the console URL"
+    )
+
+
+def post(path, request_data):
+    base, headers = _client()
+    data = json.dumps({"request_data": request_data}).encode("utf-8")
+    req = urllib.request.Request(base + path, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        if not raw:
+            return {}
+        try:
+            return json.loads(raw)
+        except ValueError:
+            raise ValueError(_not_json(r, raw))
+
+
+def csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+def to_ms(v):
+    if v in (None, ""):
+        return None
+    s = str(v)
+    if s.isdigit():
+        return int(s)
+    try:
+        return int(datetime.fromisoformat(s.replace("Z", "+00:00")).timestamp() * 1000)
+    except Exception:
+        return None
+
+
+# input name -> (API filter field, operator). Values are comma-separated lists,
+# which is the shape every one of these fields takes in the API.
+LIST_FILTERS = [
+    ("hostname", "hostname", "in"),
+    ("alias", "alias", "in"),
+    ("endpoint_id", "endpoint_id_list", "in"),
+    ("ip_list", "ip_list", "in"),
+    ("public_ip_list", "public_ip_list", "in"),
+    ("group_name", "group_name", "in"),
+    ("platform", "platform", "in"),
+    ("dist_name", "dist_name", "in"),
+    ("status", "endpoint_status", "IN"),
+    ("isolate", "isolate", "in"),
+    ("scan_status", "scan_status", "IN"),
+    ("username", "username", "IN"),
+]
+
+# input name -> (API filter field, operator). Values are epoch-ms bounds.
+RANGE_FILTERS = [
+    ("first_seen_after", "first_seen", "gte"),
+    ("first_seen_before", "first_seen", "lte"),
+    ("last_seen_after", "last_seen", "gte"),
+    ("last_seen_before", "last_seen", "lte"),
+]
+
+
+def build_filters(inputs):
+    filters = []
+    for name, field, operator in LIST_FILTERS:
+        values = csv(inputs.get(name))
+        if values:
+            filters.append({"field": field, "operator": operator, "value": values})
+    for name, field, operator in RANGE_FILTERS:
+        ms = to_ms(inputs.get(name))
+        if ms is not None:
+            filters.append({"field": field, "operator": operator, "value": ms})
+    # Escape hatch for any filter this command does not name explicitly: a JSON
+    # array of {field, operator, value} blocks, appended as-is.
+    raw = inputs.get("filters")
+    if raw:
+        try:
+            extra = json.loads(raw) if isinstance(raw, str) else raw
+        except ValueError:
+            extra = None
+        if not isinstance(extra, list) or not all(isinstance(f, dict) for f in extra):
+            raise ValueError("filters must be a JSON array of {field, operator, value} objects, got: " + str(raw)[:120])
+        filters.extend(extra)
+    return filters
+
+
+def count(filters):
+    # The endpoints grid answers with reply.total_count: the number of endpoints
+    # matching the filters, independent of the page asked for. search_to=1 keeps
+    # the page — and so the payload — to a single endpoint record.
+    request_data = {"search_from": 0, "search_to": 1, "filters": filters}
+    reply = post("/endpoints/get_endpoint/", request_data).get("reply") or {}
+    return int(reply.get("total_count") or 0)
+
+
+def main():
+    inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+    filters = build_filters(inputs)
+    result = {"count": count(filters), "filters": filters}
+    # The fleet total is the denominator the filtered count is read against: it
+    # turns a bare number into a proportion, and makes a filter that matched the
+    # whole estate — because it was too broad — visible instead of plausible.
+    result["total"] = count([]) if filters else 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)