feat(sentinelone): curated EDR integration (9 analyst commands, API v2.1)

Endpoints derived from the SentinelOne API v2.1.
Script-based (urllib, INTEGRATION_SECRETS/INPUTS contract) to express S1's nested
filter bodies. Config: console url + api_token (ApiToken header).

Commands:
- enrich: get_threats, list_agents, get_agent, get_hash_verdict
- respond: isolate_agent (disconnect), reconnect_agent (connect), mitigate_threat
  (kill/quarantine/remediate/rollback), initiate_scan, write_threat_note

integrations/sentinelone/manifest.yaml

@@ -0,0 +1,123 @@
+id: sentinelone
+name: SentinelOne
+version: 1.0.0
+description: "SentinelOne Singularity (API v2.1) — endpoint detection & response: triage threats, enrich, isolate/reconnect hosts, mitigate, scan."
+changelog: "1.0.0 — Initial release: threats, agents, hash verdict, isolate/reconnect, mitigate, scan, threat notes."
+category: endpoint
+
+# Per-instance configuration. The scripts build the API base as <url>/web/api/v2.1.
+config_schema:
+  properties:
+    url:
+      type: string
+      description: SentinelOne console URL, e.g. https://usea1.sentinelone.net
+    api_token:
+      type: string
+      description: API token (console → My User → API Token)
+      x-soar-sensitive: true
+  required:
+    - url
+    - api_token
+
+# Documented for reference; the bundled scripts build the header themselves
+# (Authorization: ApiToken <token>).
+auth:
+  - id: apitoken
+    type: api_key
+    in: header
+    name: Authorization
+    value_template: "ApiToken {{secret}}"
+    secret_field: api_token
+
+commands:
+  # ── Enrichment / read ─────────────────────────────────────────────────────
+  - id: get_threats
+    name: Get threats
+    description: List threats/detections matching filters.
+    inputs_schema:
+      properties:
+        limit: { type: number, description: "Max results (default 20)" }
+        mitigation_status: { type: string, description: "mitigated | active | blocked | suspicious | pending" }
+        query: { type: string, description: "Free-text (hash, file, computer name, uuid)" }
+        threat_ids: { type: string, description: "Comma-separated threat IDs" }
+        created_after: { type: string, description: "ISO8601 lower bound on createdAt" }
+      required: []
+    outputs_schema: { properties: {} }
+
+  - id: list_agents
+    name: List agents
+    description: List endpoints (agents) matching filters.
+    inputs_schema:
+      properties:
+        computer_name: { type: string, description: "Substring match on computer name" }
+        os_type: { type: string, description: "windows | macos | linux" }
+        is_active: { type: boolean, description: "Only active agents" }
+        limit: { type: number, description: "Max results (default 50)" }
+      required: []
+    outputs_schema: { properties: {} }
+
+  - id: get_agent
+    name: Get agent
+    description: Get details for one or more agents by ID.
+    inputs_schema:
+      properties:
+        agent_ids: { type: string, description: "Comma-separated agent IDs" }
+      required: [agent_ids]
+    outputs_schema: { properties: {} }
+
+  - id: get_hash_verdict
+    name: Get hash verdict
+    description: Reputation verdict for a SHA1 hash.
+    inputs_schema:
+      properties:
+        hash: { type: string, description: "SHA1 hash" }
+      required: [hash]
+    outputs_schema: { properties: {} }
+
+  # ── Response ──────────────────────────────────────────────────────────────
+  - id: isolate_agent
+    name: Isolate agent (disconnect)
+    description: Disconnect agents from the network.
+    inputs_schema:
+      properties:
+        agent_ids: { type: string, description: "Comma-separated agent IDs" }
+      required: [agent_ids]
+    outputs_schema: { properties: {} }
+
+  - id: reconnect_agent
+    name: Reconnect agent
+    description: Reconnect agents to the network.
+    inputs_schema:
+      properties:
+        agent_ids: { type: string, description: "Comma-separated agent IDs" }
+      required: [agent_ids]
+    outputs_schema: { properties: {} }
+
+  - id: mitigate_threat
+    name: Mitigate threat
+    description: Apply a mitigation action to threats (kill, quarantine, remediate, rollback).
+    inputs_schema:
+      properties:
+        action: { type: string, description: "kill | quarantine | un-quarantine | remediate | rollback-remediation" }
+        threat_ids: { type: string, description: "Comma-separated threat IDs" }
+      required: [action, threat_ids]
+    outputs_schema: { properties: {} }
+
+  - id: initiate_scan
+    name: Initiate endpoint scan
+    description: Start a full disk scan on agents.
+    inputs_schema:
+      properties:
+        agent_ids: { type: string, description: "Comma-separated agent IDs" }
+      required: [agent_ids]
+    outputs_schema: { properties: {} }
+
+  - id: write_threat_note
+    name: Add threat note
+    description: Add a note to one or more threats.
+    inputs_schema:
+      properties:
+        threat_ids: { type: string, description: "Comma-separated threat IDs" }
+        note: { type: string, description: "Note text" }
+      required: [threat_ids, note]
+    outputs_schema: { properties: {} }

integrations/sentinelone/scripts/get_agent.py

@@ -0,0 +1,36 @@
+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",
+    }
+
+    url = base + "/agents?" + urllib.parse.urlencode({"ids": ",".join(csv(inputs.get("agent_ids")))})
+    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)

integrations/sentinelone/scripts/get_hash_verdict.py

@@ -0,0 +1,33 @@
+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",
+    }
+
+    h = urllib.parse.quote(str(inputs.get("hash", "")), safe="")
+    url = base + "/hashes/" + h + "/verdict"
+    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)

integrations/sentinelone/scripts/get_threats.py

@@ -0,0 +1,42 @@
+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"])
+
+    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)

integrations/sentinelone/scripts/initiate_scan.py

@@ -0,0 +1,38 @@
+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",
+    }
+
+    body = {"filter": {"ids": csv(inputs.get("agent_ids"))}, "data": {}}
+    url = base + "/agents/actions/initiate-scan"
+    print(json.dumps(request("POST", url, headers, body)))
+
+
+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)

integrations/sentinelone/scripts/isolate_agent.py

@@ -0,0 +1,38 @@
+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",
+    }
+
+    body = {"filter": {"ids": csv(inputs.get("agent_ids"))}}
+    url = base + "/agents/actions/disconnect"
+    print(json.dumps(request("POST", url, headers, body)))
+
+
+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)

integrations/sentinelone/scripts/list_agents.py

@@ -0,0 +1,40 @@
+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 50)}
+    if inputs.get("computer_name"):
+        qs["computerName__like"] = str(inputs["computer_name"])
+    if inputs.get("os_type"):
+        qs["osTypes"] = str(inputs["os_type"])
+    if inputs.get("is_active") is not None:
+        qs["isActive"] = "true" if inputs["is_active"] in (True, "true", "True", 1) else "false"
+
+    url = base + "/agents?" + 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)

integrations/sentinelone/scripts/mitigate_threat.py

@@ -0,0 +1,39 @@
+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",
+    }
+
+    action = urllib.parse.quote(str(inputs.get("action", "")), safe="")
+    body = {"filter": {"ids": csv(inputs.get("threat_ids"))}}
+    url = base + "/threats/mitigate/" + action
+    print(json.dumps(request("POST", url, headers, body)))
+
+
+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)

integrations/sentinelone/scripts/reconnect_agent.py

@@ -0,0 +1,38 @@
+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",
+    }
+
+    body = {"filter": {"ids": csv(inputs.get("agent_ids"))}}
+    url = base + "/agents/actions/connect"
+    print(json.dumps(request("POST", url, headers, body)))
+
+
+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)

integrations/sentinelone/scripts/write_threat_note.py

@@ -0,0 +1,38 @@
+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",
+    }
+
+    body = {"data": {"text": str(inputs.get("note", ""))}, "filter": {"ids": csv(inputs.get("threat_ids"))}}
+    url = base + "/threats/notes"
+    print(json.dumps(request("POST", url, headers, body)))
+
+
+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)