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>
45 lines
1.6 KiB
Python
45 lines
1.6 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 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",
|
|
}
|
|
|
|
qs = {"limit": int(inputs.get("limit") or 20)}
|
|
if inputs.get("mitigation_status"):
|
|
qs["mitigationStatuses"] = str(inputs["mitigation_status"])
|
|
if inputs.get("query"):
|
|
qs["query"] = str(inputs["query"])
|
|
if inputs.get("threat_ids"):
|
|
qs["ids"] = str(inputs["threat_ids"])
|
|
if inputs.get("created_after"):
|
|
qs["createdAt__gt"] = str(inputs["created_after"])
|
|
if inputs.get("account_ids"):
|
|
qs["accountIds"] = str(inputs["account_ids"])
|
|
|
|
url = base + "/threats?" + urllib.parse.urlencode(qs)
|
|
print(json.dumps(request("GET", url, headers)))
|
|
|
|
|
|
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)
|