feat(maltiverse): new Maltiverse enrichment integration
5 commands: IP/domain/URL/file threat-intel reputation. Bearer-token auth, stdlib-only. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
id: maltiverse
|
||||
name: Maltiverse
|
||||
version: 1.0.0
|
||||
description: "Maltiverse (API) — threat-intelligence reputation for IPs, hostnames/domains, URLs and file samples (classification, blacklist sources, tags, first/last seen). Bearer-token authentication; stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: IP, hostname/domain, URL and file-sample reputation."
|
||||
category: enrichment
|
||||
|
||||
config_schema:
|
||||
properties:
|
||||
api_key:
|
||||
type: string
|
||||
description: "Maltiverse API key (Bearer token, from your Maltiverse account)"
|
||||
x-soar-sensitive: true
|
||||
required:
|
||||
- api_key
|
||||
|
||||
commands:
|
||||
- id: ip_reputation
|
||||
name: maltiverse-ip
|
||||
description: "Threat-intelligence reputation for an IP address."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
ip: { type: string, description: "IP address" }
|
||||
required: [ip]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: domain_reputation
|
||||
name: maltiverse-domain
|
||||
description: "Threat-intelligence reputation for a hostname/domain."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
domain: { type: string, description: "Hostname or domain" }
|
||||
required: [domain]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: url_reputation
|
||||
name: maltiverse-url
|
||||
description: "Threat-intelligence reputation for a URL."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
url: { type: string, description: "URL" }
|
||||
required: [url]
|
||||
outputs_schema: { properties: {} }
|
||||
- id: file_reputation
|
||||
name: maltiverse-file
|
||||
description: "Threat-intelligence reputation for a file sample by hash (MD5, SHA1 or SHA256)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
file: { type: string, description: "File hash" }
|
||||
required: [file]
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: maltiverse-test-connection
|
||||
description: "Verify the API key (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
@@ -0,0 +1,38 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.maltiverse.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path):
|
||||
url = API + path
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + str(_cfg().get("api_key") or "")}
|
||||
req = urllib.request.Request(url, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
domain = inputs.get("domain")
|
||||
if not domain:
|
||||
raise Exception("domain is required")
|
||||
result = request("GET", "/hostname/" + q(domain))
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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,38 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.maltiverse.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path):
|
||||
url = API + path
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + str(_cfg().get("api_key") or "")}
|
||||
req = urllib.request.Request(url, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
file_hash = inputs.get("file")
|
||||
if not file_hash:
|
||||
raise Exception("file is required")
|
||||
result = request("GET", "/sample/" + q(file_hash))
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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,38 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.maltiverse.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path):
|
||||
url = API + path
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + str(_cfg().get("api_key") or "")}
|
||||
req = urllib.request.Request(url, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
ip = inputs.get("ip")
|
||||
if not ip:
|
||||
raise Exception("ip is required")
|
||||
result = request("GET", "/ip/" + q(ip))
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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,36 @@
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.maltiverse.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path):
|
||||
url = API + path
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + str(_cfg().get("api_key") or "")}
|
||||
req = urllib.request.Request(url, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
result = request("GET", "/ip/8.8.8.8")
|
||||
if not isinstance(result, dict):
|
||||
raise Exception("unexpected response")
|
||||
print(json.dumps({"ok": True}))
|
||||
|
||||
|
||||
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,40 @@
|
||||
import hashlib
|
||||
import json, os, sys, urllib.parse, urllib.request, urllib.error
|
||||
|
||||
API = "https://api.maltiverse.com"
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def request(method, path):
|
||||
url = API + path
|
||||
headers = {"Accept": "application/json", "Authorization": "Bearer " + str(_cfg().get("api_key") or "")}
|
||||
req = urllib.request.Request(url, headers=headers, method=method)
|
||||
with urllib.request.urlopen(req, timeout=60) as r:
|
||||
raw = r.read()
|
||||
return json.loads(raw) if raw else {}
|
||||
|
||||
|
||||
q = lambda v: urllib.parse.quote(str(v), safe="")
|
||||
|
||||
|
||||
def main():
|
||||
inputs = json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
url = inputs.get("url")
|
||||
if not url:
|
||||
raise Exception("url is required")
|
||||
checksum = hashlib.sha256(url.encode("utf-8")).hexdigest()
|
||||
result = request("GET", "/url/" + checksum)
|
||||
print(json.dumps(result))
|
||||
|
||||
|
||||
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