riposte-marketplace/integrations/crowdstrike/scripts/device_count.py

import datetime, json, os, re, 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", "<="),
]


# A bound on first_seen / last_seen can also be relative to now ("24h", "7d",
# "30m"), so a scheduled playbook counts the sensors seen in the last day without
# computing a date. It is turned into the UTC instant here.
RELATIVE = re.compile(r"^\s*(\d+)\s*([mhd])\s*$", re.I)
UNITS = {"m": "minutes", "h": "hours", "d": "days"}


def instant(v):
    m = RELATIVE.match(str(v))
    if not m:
        return v
    delta = datetime.timedelta(**{UNITS[m.group(2).lower()]: int(m.group(1))})
    return (datetime.datetime.now(datetime.timezone.utc) - delta).strftime("%Y-%m-%dT%H:%M:%SZ")


# Sensor version bounds. FQL compares agent_version as text, where 7.9 sorts
# above 7.18, so these are compared here, number by number, over the devices the
# other filters keep.
VERSION_BOUNDS = {
    "agent_version_lt": lambda v, t: v < t,
    "agent_version_lte": lambda v, t: v <= t,
    "agent_version_gt": lambda v, t: v > t,
    "agent_version_gte": lambda v, t: v >= t,
}
SCROLL = 5000
DETAILS = 1000


def parse_version(v):
    parts = re.findall(r"\d+", str(v or ""))
    return tuple(int(x) for x in parts) if parts else None


def pad(a, b):
    n = max(len(a), len(b))
    return a + (0,) * (n - len(a)), b + (0,) * (n - len(b))


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(instant(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 version_bounds(inputs):
    bounds = {}
    for field in VERSION_BOUNDS:
        value = str(inputs.get(field) or "").strip()
        if value:
            target = parse_version(value)
            if target is None:
                raise ValueError(field + " must be a version such as 7.18.18803, got: " + repr(value))
            bounds[field] = target
    return bounds


def count_by_version(tok, fql, bounds):
    """Every device ID the filter keeps, then their sensor versions, compared
    here. A device with no readable version is left out and counted apart."""
    ids, offset = [], None
    while True:
        params = [("limit", SCROLL)]
        if fql:
            params.insert(0, ("filter", fql))
        if offset:
            params.append(("offset", offset))
        reply = call("GET", "/devices/queries/devices-scroll/v1", tok, params=params)
        page = reply.get("resources") or []
        ids.extend(page)
        pagination = (reply.get("meta") or {}).get("pagination") or {}
        offset = pagination.get("offset")
        if not page or not offset or len(ids) >= int(pagination.get("total") or 0):
            break
    matched, unreadable = 0, 0
    for i in range(0, len(ids), DETAILS):
        reply = call("POST", "/devices/entities/devices/v2", tok, body={"ids": ids[i:i + DETAILS]})
        for device in reply.get("resources") or []:
            version = parse_version(device.get("agent_version"))
            if version is None:
                unreadable += 1
            elif all(VERSION_BOUNDS[f](*pad(version, t)) for f, t in bounds.items()):
                matched += 1
    return matched, len(ids), unreadable


def main():
    tok = token()
    fql = build_fql(INPUTS)
    bounds = version_bounds(INPUTS)
    if bounds:
        matched, scanned, unreadable = count_by_version(tok, fql, bounds)
        result = {"count": matched, "filter": fql,
                  "agent_version_bounds": {f: INPUTS[f] for f in bounds},
                  "scanned": scanned, "unreadable_version": unreadable}
        result["total"] = count(tok, "")
        print(json.dumps(result))
        return
    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)