feat(crowdstrike): count devices by filter without listing them (1.4.0)

The device query already answers with meta.pagination.total — the number
of devices matching the filter, independent of the page asked for — so
device_count asks for a one-ID page and reads it, where search-device
resolves every matching device in full through
/devices/entities/devices/v2 to reach the same figure, and only filters
on four properties.

Filters cover sixteen device FQL properties (hostname, device id, local
and external IP, platform, OS version, role, containment status, sensor
version, AD domain/OU/site, serial, tags, host groups, RFM), add
first_seen/last_seen bounds, and keep a raw FQL escape hatch.

Two details the reference client leaves to chance are handled here.
Several values on one property become an OR group, parenthesised so it
binds before the '+' that ANDs it to the other clauses. And values are
quoted with the single quote escaped, so a hostname carrying an
apostrophe cannot end the literal early and change what is counted.

devices/v1 stops reporting a total on an estate it cannot page to the
end of, so a missing total falls back to the scroll query, which has no
such ceiling.

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/crowdstrike/scripts/device_count.py

@@ -0,0 +1,131 @@
+import json, os, sys, urllib.request, urllib.parse, urllib.error
+
+SECRETS = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+INPUTS = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+BASE = SECRETS.get("base_url", "https://api.crowdstrike.com").rstrip("/")
+
+
+def token():
+    data = urllib.parse.urlencode({"client_id": SECRETS.get("client_id", ""), "client_secret": SECRETS.get("client_secret", "")}).encode()
+    req = urllib.request.Request(BASE + "/oauth2/token", data=data, headers={"Content-Type": "application/x-www-form-urlencoded", "Accept": "application/json"}, method="POST")
+    with urllib.request.urlopen(req, timeout=30) as r:
+        return json.loads(r.read()).get("access_token", "")
+
+
+def call(method, path, tok, params=None, body=None):
+    url = BASE + path
+    if params:
+        # params is a list of (key, value) tuples to allow repeated keys (e.g. ids)
+        url += "?" + urllib.parse.urlencode(params)
+    data = json.dumps(body).encode() if body is not None else None
+    headers = {"Authorization": "Bearer " + tok, "Accept": "application/json"}
+    if data is not None:
+        headers["Content-Type"] = "application/json"
+    req = urllib.request.Request(url, data=data, headers=headers, method=method)
+    with urllib.request.urlopen(req, timeout=90) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def csv(v):
+    return [x.strip() for x in str(v or "").split(",") if x.strip()]
+
+
+# input name -> FQL property. Values are comma-separated; several values on one
+# property become an OR group, and the groups are ANDed together.
+STRING_FILTERS = [
+    ("hostname", "hostname"),
+    ("device_id", "device_id"),
+    ("local_ip", "local_ip"),
+    ("external_ip", "external_ip"),
+    ("platform_name", "platform_name"),
+    ("os_version", "os_version"),
+    ("product_type_desc", "product_type_desc"),
+    ("status", "status"),
+    ("agent_version", "agent_version"),
+    ("machine_domain", "machine_domain"),
+    ("ou", "ou"),
+    ("site_name", "site_name"),
+    ("serial_number", "serial_number"),
+    ("tags", "tags"),
+    ("groups", "groups"),
+    ("reduced_functionality_mode", "reduced_functionality_mode"),
+]
+
+# input name -> (FQL property, operator). Values are UTC timestamps.
+RANGE_FILTERS = [
+    ("first_seen_after", "first_seen", ">="),
+    ("first_seen_before", "first_seen", "<="),
+    ("last_seen_after", "last_seen", ">="),
+    ("last_seen_before", "last_seen", "<="),
+]
+
+
+def literal(v):
+    # FQL quotes strings with single quotes, so a value carrying one would end
+    # the literal early and change what the query means.
+    return "'" + str(v).replace("\\", "\\\\").replace("'", "\\'") + "'"
+
+
+def build_fql(inputs):
+    clauses = []
+    for name, prop in STRING_FILTERS:
+        values = csv(inputs.get(name))
+        if not values:
+            continue
+        if len(values) == 1:
+            clauses.append(prop + ":" + literal(values[0]))
+        else:
+            # ',' is OR in FQL. Parenthesised so the group binds before the '+'
+            # that ANDs it to the other clauses.
+            clauses.append("(" + ",".join(prop + ":" + literal(v) for v in values) + ")")
+    for name, prop, operator in RANGE_FILTERS:
+        value = inputs.get(name)
+        if value:
+            clauses.append(prop + ":" + operator + literal(value))
+    # Escape hatch for any property this command does not name explicitly: raw
+    # FQL, ANDed with the rest.
+    raw = str(inputs.get("filter") or "").strip()
+    if raw:
+        clauses.append("(" + raw + ")")
+    return "+".join(clauses)
+
+
+def count(tok, fql):
+    # The device query answers with meta.pagination.total: the number of devices
+    # matching the filter, independent of the page asked for. limit=1 keeps the
+    # page — and so the payload — to a single device ID.
+    params = [("limit", 1)]
+    if fql:
+        params.insert(0, ("filter", fql))
+    reply = call("GET", "/devices/queries/devices/v1", tok, params=params)
+    total = ((reply.get("meta") or {}).get("pagination") or {}).get("total")
+    if total is None:
+        # devices/v1 stops reporting a total on an estate it cannot page to the
+        # end of (it refuses past 10000). The scroll query has no such ceiling.
+        reply = call("GET", "/devices/queries/devices-scroll/v1", tok, params=params)
+        total = ((reply.get("meta") or {}).get("pagination") or {}).get("total")
+    if total is None:
+        raise ValueError("the API returned no meta.pagination.total for this query")
+    return int(total)
+
+
+def main():
+    tok = token()
+    fql = build_fql(INPUTS)
+    result = {"count": count(tok, fql), "filter": fql}
+    # The estate 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(tok, "") if fql 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)