feat: SentinelOne SDL + VirusTotal Hunting integrations
SentinelOne SDL (endpoint): Unified Alerts via the GraphQL API. Alert ingestion (get_alerts) with rich filtering and an exhaustive OCSF mapper + 'SentinelOne SDL Alert' default type, full alert details, update (status/verdict/assignee), add note and trigger mitigation action. ApiToken auth; watermark converted to epoch ms for the detectedAt filter; alert edges flattened to nodes for ingestion. VirusTotal Hunting (enrichment, Premium): Livehunt notification-file ingestion (livehunt_files) with an OCSF mapper + 'VirusTotal Hunting File' default type (severity bucketed from malicious AV detections), Livehunt notifications listing, and Retrohunt job + matching-file listing. The core VT v3 reputation already ships as 'virustotal'; the XSOAR-feed and Premium file-download/zip/pcap commands were intentionally left out. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,3 @@
|
||||
name: "VirusTotal Hunting File"
|
||||
color: "#394eff"
|
||||
icon: "alert"
|
||||
@@ -0,0 +1,93 @@
|
||||
id: virustotal_hunting
|
||||
name: VirusTotal Hunting
|
||||
version: 1.0.0
|
||||
description: "VirusTotal Hunting (Premium API v3) — ingest Livehunt notification files as findings, list Livehunt notifications, and list finished Retrohunt jobs and their matching files."
|
||||
changelog: "1.0.0 — Initial release: Livehunt notification-file ingestion (livehunt_files) with an OCSF mapper, Livehunt notifications listing, and Retrohunt job + matching-file listing."
|
||||
category: enrichment
|
||||
|
||||
# Per-instance configuration. Authenticated with a VirusTotal API key (Premium
|
||||
# subscription required for Hunting features).
|
||||
config_schema:
|
||||
properties:
|
||||
base_url:
|
||||
type: string
|
||||
description: "VirusTotal API v3 base URL"
|
||||
default: https://www.virustotal.com/api/v3
|
||||
api_key:
|
||||
type: string
|
||||
description: "VirusTotal API key (Premium)"
|
||||
x-soar-sensitive: true
|
||||
required:
|
||||
- api_key
|
||||
|
||||
auth:
|
||||
- id: apikey
|
||||
type: api_key
|
||||
in: header
|
||||
name: x-apikey
|
||||
value_template: "{{secret}}"
|
||||
secret_field: api_key
|
||||
|
||||
commands:
|
||||
# ── Ingestion ───────────────────────────────────────────────────────────────
|
||||
- id: livehunt_files
|
||||
name: vt-livehunt-files
|
||||
description: "Fetch file objects matched by Livehunt notifications. Used for ingestion: results path = data. Deduplicated by file ID (sha256)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
filter: { type: string, description: "Filter on the notification tags / ruleset name (exact name). Leave empty for all." }
|
||||
limit: { type: number, description: "Maximum number of files to fetch (default 10, max 40)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
ingest:
|
||||
results_path: data
|
||||
dedup_key: id
|
||||
|
||||
- id: livehunt_notifications
|
||||
name: vt-livehunt-notifications
|
||||
description: "List Livehunt notifications (metadata), optionally filtered by tag."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
filter: { type: string, description: "Filter on the notification tags" }
|
||||
limit: { type: number, description: "Maximum number of notifications (default 10, max 40)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: retrohunt_jobs
|
||||
name: vt-retrohunt-jobs
|
||||
description: "List Retrohunt jobs (finished by default)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
filter: { type: string, description: "Job filter (default status:finished)" }
|
||||
limit: { type: number, description: "Maximum number of jobs (default 10)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: retrohunt_matching_files
|
||||
name: vt-retrohunt-matching-files
|
||||
description: "List the files matched by a Retrohunt job."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
job_id: { type: string, description: "Retrohunt job ID" }
|
||||
limit: { type: number, description: "Maximum number of files (default 40)" }
|
||||
required: [job_id]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
# ── Connectivity test ─────────────────────────────────────────────────────
|
||||
- id: test_connection
|
||||
name: vt-hunting-test-connection
|
||||
description: "Verify connectivity and credentials (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
ingestion:
|
||||
command: livehunt_files
|
||||
mapper: livehunt_files
|
||||
default_incident_type: "VirusTotal Hunting File"
|
||||
@@ -0,0 +1,29 @@
|
||||
name: "VirusTotal Livehunt Files → OCSF"
|
||||
description: "Maps a VirusTotal Livehunt notification file (intelligence/hunting_notification_files, results_path = data) to OCSF Detection Finding fields. Each item is a file that matched a Livehunt YARA ruleset."
|
||||
field_mappings:
|
||||
title: "attributes.meaningful_name"
|
||||
# Bucket the number of malicious AV detections onto the 1-5 incident scale.
|
||||
# No parentheses (right-associative); the lowest band uses '>= 0 ? 1' so the literal sits in a then-branch.
|
||||
severity: "attributes.last_analysis_stats.malicious >= 20 ? 5 : attributes.last_analysis_stats.malicious >= 10 ? 4 : attributes.last_analysis_stats.malicious >= 5 ? 3 : attributes.last_analysis_stats.malicious >= 1 ? 2 : attributes.last_analysis_stats.malicious >= 0 ? 1 : 1"
|
||||
# results_path = data; source_path is JSONata over ONE file object.
|
||||
ocsf:
|
||||
# ── Finding ───────────────────────────────────────────────────────
|
||||
- { source_path: "id", ocsf_field: "finding_info.uid" }
|
||||
- { source_path: "attributes.meaningful_name", ocsf_field: "finding_info.title" }
|
||||
- { source_path: "context_attributes.notification_date", ocsf_field: "finding_info.created_time" }
|
||||
- { source_path: "attributes.first_submission_date", ocsf_field: "finding_info.first_seen_time" }
|
||||
- { source_path: "attributes.last_submission_date", ocsf_field: "finding_info.last_seen_time" }
|
||||
- { source_path: "attributes.reputation", ocsf_field: "risk_score" }
|
||||
- { source_path: "attributes.last_analysis_stats.malicious", ocsf_field: "count" }
|
||||
# ── Matching Livehunt rule (analytic) ─────────────────────────────
|
||||
- { source_path: "context_attributes.rule_name", ocsf_field: "finding_info.analytic.name" }
|
||||
- { source_path: "context_attributes.ruleset_name", ocsf_field: "finding_info.analytic.uid" }
|
||||
- { source_path: "context_attributes.ruleset_name", ocsf_field: "malware.name" }
|
||||
# ── Matched file ──────────────────────────────────────────────────
|
||||
- { source_path: "attributes.meaningful_name", ocsf_field: "file.name" }
|
||||
- { source_path: "attributes.type_description", ocsf_field: "file.type" }
|
||||
- { source_path: "attributes.size", ocsf_field: "file.size" }
|
||||
- { source_path: "attributes.sha256", ocsf_field: "file.hashes.sha256" }
|
||||
- { source_path: "attributes.sha1", ocsf_field: "file.hashes.sha1" }
|
||||
- { source_path: "attributes.md5", ocsf_field: "file.hashes.md5" }
|
||||
- { source_path: "attributes.ssdeep", ocsf_field: "file.hashes.ssdeep" }
|
||||
@@ -0,0 +1,38 @@
|
||||
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = str(s.get("base_url") or "https://www.virustotal.com/api/v3").rstrip("/")
|
||||
headers = {"x-apikey": s.get("api_key", ""), "Accept": "application/json"}
|
||||
return base, headers
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(path, params):
|
||||
base, headers = _cfg()
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
url = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "")
|
||||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def run():
|
||||
inp = _inputs()
|
||||
params = {"filter": inp.get("filter"), "limit": min(int(inp.get("limit") or 10), 40)}
|
||||
print(json.dumps(request("intelligence/hunting_notification_files", params)))
|
||||
|
||||
|
||||
try:
|
||||
run()
|
||||
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,38 @@
|
||||
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = str(s.get("base_url") or "https://www.virustotal.com/api/v3").rstrip("/")
|
||||
headers = {"x-apikey": s.get("api_key", ""), "Accept": "application/json"}
|
||||
return base, headers
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(path, params):
|
||||
base, headers = _cfg()
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
url = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "")
|
||||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def run():
|
||||
inp = _inputs()
|
||||
params = {"filter": inp.get("filter"), "limit": min(int(inp.get("limit") or 10), 40)}
|
||||
print(json.dumps(request("intelligence/hunting_notifications", params)))
|
||||
|
||||
|
||||
try:
|
||||
run()
|
||||
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,38 @@
|
||||
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = str(s.get("base_url") or "https://www.virustotal.com/api/v3").rstrip("/")
|
||||
headers = {"x-apikey": s.get("api_key", ""), "Accept": "application/json"}
|
||||
return base, headers
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(path, params):
|
||||
base, headers = _cfg()
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
url = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "")
|
||||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def run():
|
||||
inp = _inputs()
|
||||
params = {"filter": inp.get("filter") or "status:finished", "limit": int(inp.get("limit") or 10)}
|
||||
print(json.dumps(request("intelligence/retrohunt_jobs", params)))
|
||||
|
||||
|
||||
try:
|
||||
run()
|
||||
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,39 @@
|
||||
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = str(s.get("base_url") or "https://www.virustotal.com/api/v3").rstrip("/")
|
||||
headers = {"x-apikey": s.get("api_key", ""), "Accept": "application/json"}
|
||||
return base, headers
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def request(path, params):
|
||||
base, headers = _cfg()
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
url = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "")
|
||||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=90) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def run():
|
||||
inp = _inputs()
|
||||
job_id = urllib.parse.quote(str(inp.get("job_id", "")), safe="")
|
||||
params = {"limit": min(int(inp.get("limit") or 40), 40)}
|
||||
print(json.dumps(request("intelligence/retrohunt_jobs/" + job_id + "/matching_files", params)))
|
||||
|
||||
|
||||
try:
|
||||
run()
|
||||
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,38 @@
|
||||
import json, os, sys, urllib.request, urllib.parse, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
s = json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
base = str(s.get("base_url") or "https://www.virustotal.com/api/v3").rstrip("/")
|
||||
headers = {"x-apikey": s.get("api_key", ""), "Accept": "application/json"}
|
||||
return base, headers
|
||||
|
||||
|
||||
def request(path, params):
|
||||
base, headers = _cfg()
|
||||
clean = {k: v for k, v in params.items() if v not in (None, "")}
|
||||
url = base + "/" + path.lstrip("/") + ("?" + urllib.parse.urlencode(clean, doseq=True) if clean else "")
|
||||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
def run():
|
||||
request("intelligence/hunting_rulesets", {"limit": 1})
|
||||
print(json.dumps({"ok": True}))
|
||||
|
||||
|
||||
try:
|
||||
run()
|
||||
except urllib.error.HTTPError as e:
|
||||
detail = e.read().decode("utf-8", "replace")
|
||||
if e.code in (401, 403):
|
||||
msg = "API key is not valid or lacks Hunting (Premium) access."
|
||||
else:
|
||||
msg = "HTTP " + str(e.code)
|
||||
print(json.dumps({"ok": False, "error": msg, "detail": detail}))
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(json.dumps({"ok": False, "error": str(e)}))
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user