feat(threatfox): new ThreatFox IOC-intel integration

ThreatFox (abuse.ch) API v1, 5 commands: search IOC, recent IOCs, search hash,
tag IOCs. Auth-Key auth, stdlib-only.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

integrations/threatfox/manifest.yaml

@@ -0,0 +1,64 @@
+id: threatfox
+name: ThreatFox
+version: 1.0.0
+description: "ThreatFox by abuse.ch (API v1) — IOC intelligence: search IOCs by value, pull recent IOCs, search by file hash, and list IOCs for a tag. Auth-Key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: search IOC, recent IOCs, search hash, tag IOCs."
+category: enrichment
+
+# Per-instance configuration. abuse.ch requires an Auth-Key header.
+config_schema:
+  properties:
+    auth_key:
+      type: string
+      description: "abuse.ch Auth-Key"
+      x-soar-sensitive: true
+  required:
+    - auth_key
+
+commands:
+  - id: search_ioc
+    name: threatfox-search-ioc
+    description: "Search IOCs by value (IP, domain, URL, or hash)."
+    risk: read
+    inputs_schema:
+      properties:
+        search_term: { type: string, description: "IOC value to search" }
+      required: [search_term]
+    outputs_schema: { properties: {} }
+  - id: recent_iocs
+    name: threatfox-recent-iocs
+    description: "Get IOCs added in the last N days."
+    risk: read
+    inputs_schema:
+      properties:
+        days: { type: number, description: "Look-back days, 1-7 (default 1)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: search_hash
+    name: threatfox-search-hash
+    description: "Search IOCs by file hash."
+    risk: read
+    inputs_schema:
+      properties:
+        hash: { type: string, description: "MD5 or SHA-256 hash" }
+      required: [hash]
+    outputs_schema: { properties: {} }
+  - id: tag_iocs
+    name: threatfox-tag-iocs
+    description: "List IOCs for a given tag."
+    risk: read
+    inputs_schema:
+      properties:
+        tag: { type: string, description: "Tag (e.g. Cobalt Strike)" }
+        limit: { type: number, description: "Max IOCs (default 50)" }
+      required: [tag]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: threatfox-test-connection
+    description: "Verify the Auth-Key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/threatfox/scripts/recent_iocs.py

@@ -0,0 +1,44 @@
+import json, os, sys, urllib.request, urllib.error
+
+API = "https://threatfox-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, body):
+    data = json.dumps(body).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    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)
+
+
+def main(cfg, inputs):
+    days = inputs.get("days")
+
+    return query(cfg, {"query": "get_iocs", "days": int(days or 1)})
+
+
+_run(main)

integrations/threatfox/scripts/search_hash.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.request, urllib.error
+
+API = "https://threatfox-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, body):
+    data = json.dumps(body).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    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)
+
+
+def main(cfg, inputs):
+    hash_ = inputs.get("hash")
+    if not hash_:
+        raise Exception("hash is required")
+
+    return query(cfg, {"query": "search_hash", "hash": hash_})
+
+
+_run(main)

integrations/threatfox/scripts/search_ioc.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.request, urllib.error
+
+API = "https://threatfox-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, body):
+    data = json.dumps(body).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    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)
+
+
+def main(cfg, inputs):
+    search_term = inputs.get("search_term")
+    if not search_term:
+        raise Exception("search_term is required")
+
+    return query(cfg, {"query": "search_ioc", "search_term": search_term})
+
+
+_run(main)

integrations/threatfox/scripts/tag_iocs.py

@@ -0,0 +1,47 @@
+import json, os, sys, urllib.request, urllib.error
+
+API = "https://threatfox-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, body):
+    data = json.dumps(body).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    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)
+
+
+def main(cfg, inputs):
+    tag = inputs.get("tag")
+    if not tag:
+        raise Exception("tag is required")
+    limit = inputs.get("limit")
+
+    return query(cfg, {"query": "taginfo", "tag": tag, "limit": int(limit or 50)})
+
+
+_run(main)

integrations/threatfox/scripts/test_connection.py

@@ -0,0 +1,43 @@
+import json, os, sys, urllib.request, urllib.error
+
+API = "https://threatfox-api.abuse.ch/api/v1/"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def query(cfg, body):
+    data = json.dumps(body).encode("utf-8")
+    headers = {
+        "Auth-Key": str(cfg.get("auth_key", "")),
+        "Content-Type": "application/json",
+        "Accept": "application/json",
+    }
+    req = urllib.request.Request(API, data=data, headers=headers, method="POST")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+def _run(fn):
+    try:
+        print(json.dumps(fn(_cfg(), _inputs())))
+    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)
+
+
+def main(cfg, inputs):
+    query(cfg, {"query": "get_iocs", "days": 1})
+    return {"ok": True}
+
+
+_run(main)