From e8b2166d30344369b29d344ee3c31901fa47ed48 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Wed, 26 Aug 2026 17:15:44 +0200 Subject: [PATCH] feat(cortex-xdr): query the data lake in XQL (v1.4.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four commands reach the XQL engine: cortex-xdr-xql-query runs a query and waits for its rows, cortex-xdr-xql-start-query and cortex-xdr-xql-get-query-results split that same work in two when the query outlives a command, and cortex-xdr-xql-get-quota says what the day's hunting has already cost — an XQL query is billed by the data it scans. Three things the API makes easy to get wrong are handled here. Past 1000 rows it stops answering with results and hands over a stream id instead: gzipped JSON lines, unpacked into rows rather than dropped on the playbook as an archive it cannot read. A query carrying no limit stage is served the tenant's own default, so one is appended (100 rows, configurable) unless the query already has its own. The reference client does the same on a looser substring test, which also fires on a field whose name merely contains "limit". Waiting is a budget, not a promise: an integration command is killed at 60 seconds, so a query still running when the budget expires comes back PENDING with its execution_id instead of failing, and the results command collects the rows afterwards. Re-running the query would be charged against the quota a second time — which is why the pending answer names the command to call rather than inviting a retry. The relative window is parsed here (24 hours, 7 days, 3 months) since the scripts have no dateparser; an unreadable one is refused before the query is started, rather than silently widening to the tenant default. Co-Authored-By: Claude Opus 5 (1M context) --- integrations/cortex-xdr/manifest.yaml | 54 +++- .../scripts/xql_get_query_results.py | 142 ++++++++++ .../cortex-xdr/scripts/xql_get_quota.py | 77 ++++++ integrations/cortex-xdr/scripts/xql_query.py | 252 ++++++++++++++++++ .../cortex-xdr/scripts/xql_start_query.py | 165 ++++++++++++ 5 files changed, 687 insertions(+), 3 deletions(-) create mode 100644 integrations/cortex-xdr/scripts/xql_get_query_results.py create mode 100644 integrations/cortex-xdr/scripts/xql_get_quota.py create mode 100644 integrations/cortex-xdr/scripts/xql_query.py create mode 100644 integrations/cortex-xdr/scripts/xql_start_query.py diff --git a/integrations/cortex-xdr/manifest.yaml b/integrations/cortex-xdr/manifest.yaml index 42a1c22..2f292d7 100644 --- a/integrations/cortex-xdr/manifest.yaml +++ b/integrations/cortex-xdr/manifest.yaml @@ -1,8 +1,8 @@ id: cortex_xdr name: Cortex XDR -version: 1.3.2 -description: "Palo Alto Cortex XDR (public API v1) — incident and alert ingestion + write-back, endpoint isolation/scan/delete/tagging, RTR scripts, hash block/allow lists, file quarantine/restore/retrieval, alert exclusions, external alert push (parsed/CEF), device-control violations, audits, distributions and RBAC/risk." -changelog: "1.3.2 — The mappers fill the Actor group, as the HarfangLab and SentinelOne mappers already did. XDR names the process that acted actor_process_* and its account actor_effective_username, but both landed outside the actor: the process in process.*, the account in user.name — the OCSF slot for the user an event was done TO. So a Cortex tenant answered nothing to a correlation or pre-processing rule written on actor.user.name / actor.process.name, and any rule reading user.name as the victim read the attacker instead. actor_process_* now maps to actor.process.*, causality_actor_process_* to actor.process.parent_process.*, actor_effective_username to actor.user.name; process.* is left free for the process acted upon, which is what it holds in the SentinelOne (targetProcessInfo) and Carbon Black (childproc_*) mappers. On get_incidents_full the incident's impacted users[] and the first alert's user_name both wrote user.name, so the alert's actor overwrote the impacted user; they are now two fields. Rules and columns written against the old Cortex paths need repointing. 1.3.1 — get_incidents_full can watermark on modification_time (modified_after), which is what ingestion wants: an XDR incident keeps growing after creation, and a creation_time watermark fetches it once and never looks again, so every alert that joins it afterwards is lost. The full-incident fetch command is named cortex-xdr-get-incidents-full (id get_incidents_full), not cortex-xdr-fetch-incidents: sitting next to cortex-xdr-get-incidents in the command list, it now reads as what it is — the same call, everything included. A rule created against the old id must be pointed at the new one. 1.3.0 — Richer incident ingestion (get_incidents_full, on get_multiple_incidents_extra_data): incidents now arrive with their alerts in full and their file/network artifacts, where get_incidents only ever answered a 21-field summary carrying neither hosts, users, MITRE nor a single alert. Alert ingestion, and the alerts endpoint answers at last: get_alerts was sending the incidents dialect ({filters, search_from, search_to, sort}) to a grid endpoint that speaks request_data.filter_data (SEARCH_FIELD/SEARCH_TYPE/SEARCH_VALUE blocks, paging.from/to, sort as a list), and every call came back HTTP 500. Body rebuilt from the reference client, rows unwrapped out of their alert_fields envelope, severity code and the dotted status.progress key derived into readable fields. Alert ingestion: get_alerts is now a fetch command (results path reply.alerts, dedup on alert_id, incremental on source_insert_ts) with a bundled OCSF mapper and a Cortex XDR Alert incident type, so detections can be ingested alongside — or instead of — incidents. The incident mapper is fixed on the way past: its severity expression compared strings, which the mapping engine cannot do, so every ingested incident silently took the rule's default severity; it also now carries a detection anchor so MTTD is measurable. It pages past the API's 100-results-per-call ceiling, and sorts oldest-first on an incremental fetch so a truncated window drops the alerts the next poll can still see. 1.2.1 — Connection troubleshooting: the URL is normalised to the tenant host (a pasted /public_api/v1 or console path no longer breaks the call), a non-JSON reply reports the status, content type and body instead of a bare JSON parse error, missing key/key ID is caught up front, nonce and timestamp are sent in both auth modes as the reference client does, and test_connection now probes get_incidents. 1.2.0 — Incident write-back (update_incident: status/severity/assignment/resolve comment) and external alert push (insert_parsed_alerts, insert_cef_alerts). 1.1.0 — Full command coverage: added delete/alias/tag endpoints, abort scan, original alerts, script metadata/code/snippet/exec-status, file retrieval (+details), alert exclusions, device-control violations, audits, distribution url/status/create and RBAC (users, roles, groups, risk score, risky users/hosts). 1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication." +version: 1.4.0 +description: "Palo Alto Cortex XDR (public API v1) — incident and alert ingestion + write-back, endpoint isolation/scan/delete/tagging, XQL queries against the data lake, RTR scripts, hash block/allow lists, file quarantine/restore/retrieval, alert exclusions, external alert push (parsed/CEF), device-control violations, audits, distributions and RBAC/risk." +changelog: "1.4.0 — XQL. Four commands reach the data lake: cortex-xdr-xql-query runs a query and waits for its rows, cortex-xdr-xql-start-query and cortex-xdr-xql-get-query-results split that work in two when the query outlives a command, and cortex-xdr-xql-get-quota says what the day’s hunting has already cost — an XQL query is billed by the data it scans. Two things the API makes easy to get wrong are handled here. Past 1000 rows it stops answering with results and hands over a stream id instead: gzipped JSON lines, unpacked into rows rather than dropped on the playbook as an archive. And a query carrying no limit stage is served the tenant’s own default, so one is appended (100 rows, configurable) unless the query already has its own. Waiting is a budget, not a promise: a command is killed at 60 seconds, so a query still running comes back PENDING with its execution_id, which the results command collects afterwards — re-running the query would be charged against the quota a second time. 1.3.2 — The mappers fill the Actor group, as the HarfangLab and SentinelOne mappers already did. XDR names the process that acted actor_process_* and its account actor_effective_username, but both landed outside the actor: the process in process.*, the account in user.name — the OCSF slot for the user an event was done TO. So a Cortex tenant answered nothing to a correlation or pre-processing rule written on actor.user.name / actor.process.name, and any rule reading user.name as the victim read the attacker instead. actor_process_* now maps to actor.process.*, causality_actor_process_* to actor.process.parent_process.*, actor_effective_username to actor.user.name; process.* is left free for the process acted upon, which is what it holds in the SentinelOne (targetProcessInfo) and Carbon Black (childproc_*) mappers. On get_incidents_full the incident's impacted users[] and the first alert's user_name both wrote user.name, so the alert's actor overwrote the impacted user; they are now two fields. Rules and columns written against the old Cortex paths need repointing. 1.3.1 — get_incidents_full can watermark on modification_time (modified_after), which is what ingestion wants: an XDR incident keeps growing after creation, and a creation_time watermark fetches it once and never looks again, so every alert that joins it afterwards is lost. The full-incident fetch command is named cortex-xdr-get-incidents-full (id get_incidents_full), not cortex-xdr-fetch-incidents: sitting next to cortex-xdr-get-incidents in the command list, it now reads as what it is — the same call, everything included. A rule created against the old id must be pointed at the new one. 1.3.0 — Richer incident ingestion (get_incidents_full, on get_multiple_incidents_extra_data): incidents now arrive with their alerts in full and their file/network artifacts, where get_incidents only ever answered a 21-field summary carrying neither hosts, users, MITRE nor a single alert. Alert ingestion, and the alerts endpoint answers at last: get_alerts was sending the incidents dialect ({filters, search_from, search_to, sort}) to a grid endpoint that speaks request_data.filter_data (SEARCH_FIELD/SEARCH_TYPE/SEARCH_VALUE blocks, paging.from/to, sort as a list), and every call came back HTTP 500. Body rebuilt from the reference client, rows unwrapped out of their alert_fields envelope, severity code and the dotted status.progress key derived into readable fields. Alert ingestion: get_alerts is now a fetch command (results path reply.alerts, dedup on alert_id, incremental on source_insert_ts) with a bundled OCSF mapper and a Cortex XDR Alert incident type, so detections can be ingested alongside — or instead of — incidents. The incident mapper is fixed on the way past: its severity expression compared strings, which the mapping engine cannot do, so every ingested incident silently took the rule's default severity; it also now carries a detection anchor so MTTD is measurable. It pages past the API's 100-results-per-call ceiling, and sorts oldest-first on an incremental fetch so a truncated window drops the alerts the next poll can still see. 1.2.1 — Connection troubleshooting: the URL is normalised to the tenant host (a pasted /public_api/v1 or console path no longer breaks the call), a non-JSON reply reports the status, content type and body instead of a bare JSON parse error, missing key/key ID is caught up front, nonce and timestamp are sent in both auth modes as the reference client does, and test_connection now probes get_incidents. 1.2.0 — Incident write-back (update_incident: status/severity/assignment/resolve comment) and external alert push (insert_parsed_alerts, insert_cef_alerts). 1.1.0 — Full command coverage: added delete/alias/tag endpoints, abort scan, original alerts, script metadata/code/snippet/exec-status, file retrieval (+details), alert exclusions, device-control violations, audits, distribution url/status/create and RBAC (users, roles, groups, risk score, risky users/hosts). 1.0.0 — Initial release: incident ingestion (get_incidents) with OCSF mapper, endpoints, isolate/unisolate, scan, hash blocklist/allowlist, quarantine/restore, run script + results, alerts retrieval, distributions and action status. Standard or Advanced API authentication." category: endpoint # Per-instance configuration. The base URL is the tenant API root, e.g. @@ -294,6 +294,54 @@ commands: required: [] outputs_schema: { properties: {} } + # ── XQL search ──────────────────────────────────────────────────────────── + - id: xql_query + name: cortex-xdr-xql-query + description: "Run an XQL query and wait for its rows: the query is started, polled until the tenant answers, and the rows returned — including past the API's 1000-row ceiling, where results arrive as a gzipped stream this command unpacks. A command is killed at 60 seconds, so a query still running when the wait budget expires comes back with status PENDING and its execution_id: hand that to cortex-xdr-xql-get-query-results instead of running the query again, which is charged against the XQL quota a second time. Returns {reply:{status, execution_id, result_count, results:[...], query_cost, remaining_quota}}." + risk: read + inputs_schema: + properties: + query: { type: string, description: "XQL query, e.g. dataset = xdr_data | filter event_type = ENUM.PROCESS | fields agent_hostname, action_process_image_name" } + time_frame: { type: string, description: "Relative window, written as : 30 minutes, 24 hours, 7 days, 2 weeks, 3 months. Left empty, and with no from_time/to_time, the tenant applies its own default — the last 24 hours, not all of time." } + from_time: { type: string, description: "Absolute lower bound, ISO8601 or epoch ms. Wins over time_frame; on its own, the window runs up to now." } + to_time: { type: string, description: "Absolute upper bound, ISO8601 or epoch ms." } + limit: { type: number, description: "Row ceiling, appended to the query as a limit stage when it carries none of its own (default 100) — without one the tenant serves its own default. Past 1000 rows the results come back as a gzipped stream, unpacked here, so raise this knowingly: every row is held in memory." } + tenant_ids: { type: string, description: "Comma-separated tenant IDs to query, on a multi-tenant (MSSP) deployment. Empty queries the tenant the API key belongs to." } + wait: { type: number, description: "Seconds to wait for the rows before answering PENDING (default 45, capped at 55 — the command itself is killed at 60)." } + required: [query] + outputs_schema: { properties: {} } + - id: xql_start_query + name: cortex-xdr-xql-start-query + description: "Start an XQL query and return its execution_id without waiting for the rows — for a query too slow for the 60-second command ceiling, or a playbook with something else to do while the tenant works. The rows are collected afterwards with cortex-xdr-xql-get-query-results. Returns {reply:{execution_id, query, status}}." + risk: read + inputs_schema: + properties: + query: { type: string, description: "XQL query" } + time_frame: { type: string, description: "Relative window, written as : 30 minutes, 24 hours, 7 days, 2 weeks, 3 months. Empty leaves the tenant's default (the last 24 hours)." } + from_time: { type: string, description: "Absolute lower bound, ISO8601 or epoch ms. Wins over time_frame." } + to_time: { type: string, description: "Absolute upper bound, ISO8601 or epoch ms." } + limit: { type: number, description: "Row ceiling, appended as a limit stage when the query carries none (default 100)." } + tenant_ids: { type: string, description: "Comma-separated tenant IDs (multi-tenant deployments)" } + required: [query] + outputs_schema: { properties: {} } + - id: xql_get_query_results + name: cortex-xdr-xql-get-query-results + description: "Collect the rows of an XQL query started earlier. A query still running answers status PENDING with no rows, which is not a failure — poll again. Past 1000 rows the API hands over a stream id rather than the rows; the gzipped JSON lines behind it are unpacked here. Returns {reply:{status, result_count, results:[...]}}." + risk: read + inputs_schema: + properties: + execution_id: { type: string, description: "Execution ID returned by cortex-xdr-xql-start-query (the API calls it query_id — either input name is accepted)" } + required: [execution_id] + outputs_schema: { properties: {} } + - id: xql_get_quota + name: cortex-xdr-xql-get-quota + description: "Read the XQL quota: what the licence allows, what was purchased on top of it, and what the last 24 hours have consumed. A query is billed by the data it scans, so this is what says whether a hunting playbook can still run today." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + # ── Actions / misc ──────────────────────────────────────────────────────── - id: action_status name: cortex-xdr-action-status diff --git a/integrations/cortex-xdr/scripts/xql_get_query_results.py b/integrations/cortex-xdr/scripts/xql_get_query_results.py new file mode 100644 index 0000000..2ebd0eb --- /dev/null +++ b/integrations/cortex-xdr/scripts/xql_get_query_results.py @@ -0,0 +1,142 @@ +import json, os, sys, gzip, time, hashlib, secrets, string, urllib.request, urllib.error + + +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-.xdr..paloaltonetworks.com), not the console URL" + ) + + +def post_raw(path, request_data, timeout=90): + 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=timeout) as r: + return r.read(), r + + +def post(path, request_data, timeout=90): + raw, r = post_raw(path, request_data, timeout) + if not raw: + return {} + try: + return json.loads(raw) + except ValueError: + raise ValueError(_not_json(r, raw)) + + +def _error_text(reply): + err = reply.get("error") or reply.get("error_message") or "unknown error" + if isinstance(err, dict): + return "; ".join(str(k) + ": " + str(v) for k, v in err.items()) + return str(err) + + +def read_stream(stream_id, timeout=90): + """Past 1000 rows the API hands over a stream id instead of the rows. + + What comes back is gzipped JSON lines, not JSON — one object per line — so it + is unpacked here rather than handed to the caller as an opaque archive. + """ + raw, _ = post_raw("/xql/get_query_results_stream/", {"stream_id": stream_id, "is_gzip_compressed": True}, timeout) + if raw[:2] == b"\x1f\x8b": + raw = gzip.decompress(raw) + rows = [] + for line in raw.decode("utf-8", "replace").splitlines(): + line = line.strip() + if line: + rows.append(json.loads(line)) + return rows + + +def fetch_results(execution_id, timeout=90): + reply = (post("/xql/get_query_results/", { + "query_id": execution_id, + "pending_flag": True, # answer PENDING instead of holding the connection open + "format": "json", + }, timeout) or {}).get("reply") or {} + status = str(reply.get("status") or "") + if status == "FAIL": + raise ValueError("query " + execution_id + " failed: " + _error_text(reply)) + out = { + "execution_id": execution_id, + "status": status, + "number_of_results": reply.get("number_of_results"), + "query_cost": reply.get("query_cost"), + "remaining_quota": reply.get("remaining_quota"), + "results": [], + "result_count": 0, + } + if status != "SUCCESS": + return out + results = reply.get("results") + if isinstance(results, dict): + rows = read_stream(results["stream_id"], timeout) if results.get("stream_id") else (results.get("data") or []) + else: + rows = results or [] + out["results"] = rows + out["result_count"] = len(rows) + total = out.get("number_of_results") + if isinstance(total, int): + # Say it when the query matched more than it handed back: the missing + # rows are not coming, the query has to be narrowed or its limit raised. + out["truncated"] = total > len(rows) + return out + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + # The start command calls it execution_id, the API calls it query_id — both + # are accepted here so a playbook can wire either one through. + execution_id = str(inputs.get("execution_id") or inputs.get("query_id") or "").strip() + if not execution_id: + raise ValueError("execution_id is required — take it from cortex-xdr-xql-start-query") + print(json.dumps({"reply": fetch_results(execution_id)})) + + +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) diff --git a/integrations/cortex-xdr/scripts/xql_get_quota.py b/integrations/cortex-xdr/scripts/xql_get_quota.py new file mode 100644 index 0000000..c09d459 --- /dev/null +++ b/integrations/cortex-xdr/scripts/xql_get_quota.py @@ -0,0 +1,77 @@ +import json, os, sys, time, hashlib, secrets, string, urllib.request, urllib.error + + +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-.xdr..paloaltonetworks.com), not the console URL" + ) + + +def post_raw(path, request_data, timeout=90): + 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=timeout) as r: + return r.read(), r + + +def post(path, request_data, timeout=90): + raw, r = post_raw(path, request_data, timeout) + if not raw: + return {} + try: + return json.loads(raw) + except ValueError: + raise ValueError(_not_json(r, raw)) + + +def main(): + print(json.dumps(post("/xql/get_quota/", {}))) + + +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) diff --git a/integrations/cortex-xdr/scripts/xql_query.py b/integrations/cortex-xdr/scripts/xql_query.py new file mode 100644 index 0000000..09a6de0 --- /dev/null +++ b/integrations/cortex-xdr/scripts/xql_query.py @@ -0,0 +1,252 @@ +import json, os, re, sys, gzip, 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-.xdr..paloaltonetworks.com), not the console URL" + ) + + +def post_raw(path, request_data, timeout=90): + 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=timeout) as r: + return r.read(), r + + +def post(path, request_data, timeout=90): + raw, r = post_raw(path, request_data, timeout) + 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 + + +UNITS = { + "s": 1, "sec": 1, "secs": 1, "second": 1, "seconds": 1, + "m": 60, "min": 60, "mins": 60, "minute": 60, "minutes": 60, + "h": 3600, "hr": 3600, "hrs": 3600, "hour": 3600, "hours": 3600, + "d": 86400, "day": 86400, "days": 86400, + "w": 604800, "week": 604800, "weeks": 604800, + "mo": 2592000, "month": 2592000, "months": 2592000, + "y": 31536000, "year": 31536000, "years": 31536000, +} +RELATIVE = re.compile(r"^\s*(\d+)\s*([a-z]+)\s*(?:ago)?\s*$", re.I) + + +def timeframe(inputs): + """The window the query reads, in the shape the API expects. + + Absolute bounds win over the relative window; a lone from_time runs up to + now. Nothing at all leaves the timeframe out, and the tenant applies its own + default (the last 24 hours) — which is not the same as querying all of time. + """ + frm, to = to_ms(inputs.get("from_time")), to_ms(inputs.get("to_time")) + if frm is not None or to is not None: + return {"from": frm if frm is not None else 0, + "to": to if to is not None else int(time.time() * 1000)} + raw = str(inputs.get("time_frame") or "").strip() + if not raw: + return None + m = RELATIVE.match(raw) + unit = UNITS.get(m.group(2).lower()) if m else None + if not unit: + raise ValueError( + "time_frame '" + raw + "' is not a window this command can read — write it as " + " (30 minutes, 24 hours, 7 days, 2 weeks, 3 months), or give " + "from_time/to_time for absolute bounds" + ) + return {"relativeTime": int(m.group(1)) * unit * 1000} + + +# Anything already carrying its own `| limit` stage is left alone; the rest gets +# one appended. Without it the tenant returns its own default row count and the +# query is billed for a window nobody asked for (the reference client does the +# same, on a looser substring test that also fires on a field named *limit*). +LIMIT_STAGE = re.compile(r"\|\s*limit\s+\d+", re.I) + + +def build_query(inputs): + query = str(inputs.get("query") or "").strip() + if not query: + raise ValueError("query is required") + if not LIMIT_STAGE.search(query): + query += "\n| limit " + str(int(inputs.get("limit") or 100)) + return query + + +def start_query(inputs, timeout=90): + query = build_query(inputs) + rd = {"query": query} + tf = timeframe(inputs) + if tf: + rd["timeframe"] = tf + tenants = csv(inputs.get("tenant_ids")) + if tenants: + rd["tenants"] = tenants + reply = (post("/xql/start_xql_query/", rd, timeout) or {}).get("reply") + execution_id = str(reply or "").strip() + if not execution_id: + raise ValueError("the tenant accepted the query but returned no execution id") + return execution_id, query + + +def _error_text(reply): + err = reply.get("error") or reply.get("error_message") or "unknown error" + if isinstance(err, dict): + return "; ".join(str(k) + ": " + str(v) for k, v in err.items()) + return str(err) + + +def read_stream(stream_id, timeout=90): + """Past 1000 rows the API hands over a stream id instead of the rows. + + What comes back is gzipped JSON lines, not JSON — one object per line — so it + is unpacked here rather than handed to the caller as an opaque archive. + """ + raw, _ = post_raw("/xql/get_query_results_stream/", {"stream_id": stream_id, "is_gzip_compressed": True}, timeout) + if raw[:2] == b"\x1f\x8b": + raw = gzip.decompress(raw) + rows = [] + for line in raw.decode("utf-8", "replace").splitlines(): + line = line.strip() + if line: + rows.append(json.loads(line)) + return rows + + +def fetch_results(execution_id, timeout=90): + reply = (post("/xql/get_query_results/", { + "query_id": execution_id, + "pending_flag": True, # answer PENDING instead of holding the connection open + "format": "json", + }, timeout) or {}).get("reply") or {} + status = str(reply.get("status") or "") + if status == "FAIL": + raise ValueError("query " + execution_id + " failed: " + _error_text(reply)) + out = { + "execution_id": execution_id, + "status": status, + "number_of_results": reply.get("number_of_results"), + "query_cost": reply.get("query_cost"), + "remaining_quota": reply.get("remaining_quota"), + "results": [], + "result_count": 0, + } + if status != "SUCCESS": + return out + results = reply.get("results") + if isinstance(results, dict): + rows = read_stream(results["stream_id"], timeout) if results.get("stream_id") else (results.get("data") or []) + else: + rows = results or [] + out["results"] = rows + out["result_count"] = len(rows) + total = out.get("number_of_results") + if isinstance(total, int): + # Say it when the query matched more than it handed back: the missing + # rows are not coming, the query has to be narrowed or its limit raised. + out["truncated"] = total > len(rows) + return out + + +# A command is killed at 60 seconds, whatever this script would rather do. So a +# query that outlives the wait budget is not an error and not a lost query: it +# comes back PENDING with its execution id, and cortex-xdr-xql-get-query-results +# collects the rows afterwards. Re-running the query instead would be charged a +# second time against the tenant's XQL quota. +MAX_WAIT = 55 +POLL_INTERVAL = 3 +POLL_TIMEOUT = 20 + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + wait = min(MAX_WAIT, max(1, int(inputs.get("wait") or 45))) + execution_id, query = start_query(inputs, POLL_TIMEOUT) + deadline = time.time() + wait + # Look once before sleeping: a narrow query over a short window is often + # already answered by the time the start call returns. + out = fetch_results(execution_id, POLL_TIMEOUT) + while out["status"] == "PENDING": + left = deadline - time.time() + if left <= 0: + break + time.sleep(min(POLL_INTERVAL, left)) + out = fetch_results(execution_id, POLL_TIMEOUT) + out["query"] = query + if out["status"] == "PENDING": + out["hint"] = ( + "still running after " + str(wait) + "s — pass this execution_id to " + "cortex-xdr-xql-get-query-results rather than running the query again, " + "which would be billed a second time" + ) + print(json.dumps({"reply": out})) + + +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) diff --git a/integrations/cortex-xdr/scripts/xql_start_query.py b/integrations/cortex-xdr/scripts/xql_start_query.py new file mode 100644 index 0000000..76058a0 --- /dev/null +++ b/integrations/cortex-xdr/scripts/xql_start_query.py @@ -0,0 +1,165 @@ +import json, os, re, 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-.xdr..paloaltonetworks.com), not the console URL" + ) + + +def post_raw(path, request_data, timeout=90): + 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=timeout) as r: + return r.read(), r + + +def post(path, request_data, timeout=90): + raw, r = post_raw(path, request_data, timeout) + 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 + + +UNITS = { + "s": 1, "sec": 1, "secs": 1, "second": 1, "seconds": 1, + "m": 60, "min": 60, "mins": 60, "minute": 60, "minutes": 60, + "h": 3600, "hr": 3600, "hrs": 3600, "hour": 3600, "hours": 3600, + "d": 86400, "day": 86400, "days": 86400, + "w": 604800, "week": 604800, "weeks": 604800, + "mo": 2592000, "month": 2592000, "months": 2592000, + "y": 31536000, "year": 31536000, "years": 31536000, +} +RELATIVE = re.compile(r"^\s*(\d+)\s*([a-z]+)\s*(?:ago)?\s*$", re.I) + + +def timeframe(inputs): + """The window the query reads, in the shape the API expects. + + Absolute bounds win over the relative window; a lone from_time runs up to + now. Nothing at all leaves the timeframe out, and the tenant applies its own + default (the last 24 hours) — which is not the same as querying all of time. + """ + frm, to = to_ms(inputs.get("from_time")), to_ms(inputs.get("to_time")) + if frm is not None or to is not None: + return {"from": frm if frm is not None else 0, + "to": to if to is not None else int(time.time() * 1000)} + raw = str(inputs.get("time_frame") or "").strip() + if not raw: + return None + m = RELATIVE.match(raw) + unit = UNITS.get(m.group(2).lower()) if m else None + if not unit: + raise ValueError( + "time_frame '" + raw + "' is not a window this command can read — write it as " + " (30 minutes, 24 hours, 7 days, 2 weeks, 3 months), or give " + "from_time/to_time for absolute bounds" + ) + return {"relativeTime": int(m.group(1)) * unit * 1000} + + +# Anything already carrying its own `| limit` stage is left alone; the rest gets +# one appended. Without it the tenant returns its own default row count and the +# query is billed for a window nobody asked for (the reference client does the +# same, on a looser substring test that also fires on a field named *limit*). +LIMIT_STAGE = re.compile(r"\|\s*limit\s+\d+", re.I) + + +def build_query(inputs): + query = str(inputs.get("query") or "").strip() + if not query: + raise ValueError("query is required") + if not LIMIT_STAGE.search(query): + query += "\n| limit " + str(int(inputs.get("limit") or 100)) + return query + + +def start_query(inputs, timeout=90): + query = build_query(inputs) + rd = {"query": query} + tf = timeframe(inputs) + if tf: + rd["timeframe"] = tf + tenants = csv(inputs.get("tenant_ids")) + if tenants: + rd["tenants"] = tenants + reply = (post("/xql/start_xql_query/", rd, timeout) or {}).get("reply") + execution_id = str(reply or "").strip() + if not execution_id: + raise ValueError("the tenant accepted the query but returned no execution id") + return execution_id, query + + +def main(): + inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + execution_id, query = start_query(inputs) + print(json.dumps({"reply": {"execution_id": execution_id, "query": query, "status": "PENDING"}})) + + +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)