feat(cortex-xdr): query the data lake in XQL (v1.4.0)
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) <noreply@anthropic.com>
This commit is contained in:
File diff suppressed because one or more lines are too long
@@ -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-<tenant>.xdr.<region>.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)
|
||||
@@ -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-<tenant>.xdr.<region>.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)
|
||||
@@ -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-<tenant>.xdr.<region>.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 "
|
||||
"<number> <unit> (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)
|
||||
@@ -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-<tenant>.xdr.<region>.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 "
|
||||
"<number> <unit> (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)
|
||||
Reference in New Issue
Block a user