Guillaume BOURGEOIS
7efc873867
A multi-tenant console returns every account's threats and alerts at once. get_threats and get_alerts now take an optional account_ids input, passed through as the accountIds query parameter, so an ingestion can be pinned to the accounts the SOC actually watches. The input is optional, so existing instances and running ingestions keep their current behaviour. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
50 lines
1.8 KiB
Python
50 lines
1.8 KiB
Python
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
|
|
|
|
|
def request(method, url, headers, body=None):
|
|
data = json.dumps(body).encode("utf-8") if body is not None else None
|
|
req = urllib.request.Request(url, data=data, headers=headers, method=method)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
raw = resp.read()
|
|
return json.loads(raw) if raw else {}
|
|
|
|
|
|
def csv(v):
|
|
return [x.strip() for x in str(v or "").split(",") if x.strip()]
|
|
|
|
|
|
def main():
|
|
secrets = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
|
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
|
base = secrets.get("url", "").rstrip("/") + "/web/api/v2.1"
|
|
headers = {
|
|
"Authorization": "ApiToken " + secrets.get("api_token", ""),
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
}
|
|
# === REQUEST ===
|
|
qs = {
|
|
"createdAt__gte": inputs.get("created_from"),
|
|
"createdAt__lte": inputs.get("created_until"),
|
|
"ruleName__contains": inputs.get("ruleName"),
|
|
"incidentStatus": inputs.get("incidentStatus"),
|
|
"analystVerdict": inputs.get("analystVerdict"),
|
|
"ids": inputs.get("alert_ids"),
|
|
"siteIds": inputs.get("site_ids"),
|
|
"accountIds": inputs.get("account_ids"),
|
|
"limit": int(inputs.get("limit") or 100),
|
|
}
|
|
url = base + "/cloud-detection/alerts?" + urllib.parse.urlencode({k: v for k, v in qs.items() if v not in (None, "")})
|
|
print(json.dumps(request("GET", url, headers)))
|
|
# === END ===
|
|
|
|
|
|
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)
|