feat(feed-plaintext): new generic plain-text threat-intel feed connector
Fetches a line-per-indicator blocklist URL and emits normalized IOCs {value,type}
for TIM import. Comment-char/field-index/type config, auto type-detection.
Optional bearer auth, stdlib-only. py_compile clean.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Guillaume BOURGEOIS
@@ -0,0 +1,46 @@
|
||||
id: feed_plaintext
|
||||
name: Plain Text Feed
|
||||
version: 1.0.0
|
||||
description: "Generic plain-text threat-intel feed connector — fetch a line-per-indicator list from a URL (e.g. an IP or domain blocklist) and emit normalized IOCs (value + type) for import into the Threat Indicator Manager. One connector, many feeds. No authentication (or an optional bearer token); stdlib-only, no extra Python dependencies."
|
||||
changelog: "1.0.0 — Initial release: fetch indicators from a plain-text feed URL."
|
||||
category: feed
|
||||
|
||||
# Per-instance configuration. The feed is fetched over HTTP(S) from feed_url.
|
||||
config_schema:
|
||||
properties:
|
||||
feed_url:
|
||||
type: string
|
||||
description: "URL of the plain-text feed (one indicator per line)"
|
||||
api_token:
|
||||
type: string
|
||||
description: "Optional bearer token (if the feed requires auth)"
|
||||
x-soar-sensitive: true
|
||||
insecure:
|
||||
type: boolean
|
||||
description: "Trust any TLS certificate (not secure)"
|
||||
default: false
|
||||
required:
|
||||
- feed_url
|
||||
|
||||
commands:
|
||||
- id: fetch_indicators
|
||||
name: feed-plaintext-fetch-indicators
|
||||
description: "Fetch and parse the plain-text feed, returning normalized indicators."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties:
|
||||
ioc_type: { type: string, description: "auto | ip | domain | url | hash | email (default auto — detect per value)" }
|
||||
comment_char: { type: string, description: "Lines starting with this are ignored (default '#')" }
|
||||
field_index: { type: number, description: "If lines have multiple whitespace-separated fields, the 0-based field to take (default 0)" }
|
||||
max_indicators: { type: number, description: "Max indicators to return (default 10000)" }
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
|
||||
- id: test_connection
|
||||
name: feed-plaintext-test-connection
|
||||
description: "Verify the feed URL is reachable (used by the Test button)."
|
||||
risk: read
|
||||
inputs_schema:
|
||||
properties: {}
|
||||
required: []
|
||||
outputs_schema: { properties: {} }
|
||||
@@ -0,0 +1,92 @@
|
||||
import json, os, sys, re, ssl, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _fetch(cfg):
|
||||
url = str(cfg.get("feed_url", ""))
|
||||
if not url:
|
||||
raise Exception("feed_url is not configured")
|
||||
headers = {"Accept": "text/plain, */*", "User-Agent": "Riposte-SOAR"}
|
||||
if cfg.get("api_token"):
|
||||
headers["Authorization"] = "Bearer " + str(cfg["api_token"])
|
||||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=120, context=_ctx(cfg)) as r:
|
||||
return r.read().decode("utf-8", "replace")
|
||||
|
||||
|
||||
_IPV4 = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}(?:/\d{1,2})?$")
|
||||
_HASH = re.compile(r"^[a-fA-F0-9]{32}$|^[a-fA-F0-9]{40}$|^[a-fA-F0-9]{64}$")
|
||||
_EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||||
|
||||
|
||||
def detect_type(value):
|
||||
v = value.strip()
|
||||
if _IPV4.match(v) or (":" in v and re.match(r"^[0-9a-fA-F:]+$", v)):
|
||||
return "ip"
|
||||
if v.lower().startswith("http://") or v.lower().startswith("https://"):
|
||||
return "url"
|
||||
if _EMAIL.match(v):
|
||||
return "email"
|
||||
if _HASH.match(v):
|
||||
return "hash"
|
||||
if "." in v and " " not in v:
|
||||
return "domain"
|
||||
return "unknown"
|
||||
|
||||
|
||||
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):
|
||||
text = _fetch(cfg)
|
||||
cc = inputs.get("comment_char") or "#"
|
||||
itype = inputs.get("ioc_type") or "auto"
|
||||
fi = int(inputs.get("field_index") or 0)
|
||||
cap = int(inputs.get("max_indicators") or 10000)
|
||||
|
||||
indicators = []
|
||||
for line in text.splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith(cc):
|
||||
continue
|
||||
fields = line.split()
|
||||
if len(fields) <= fi:
|
||||
continue
|
||||
val = fields[fi].strip()
|
||||
if not val:
|
||||
continue
|
||||
type_ = itype if itype != "auto" else detect_type(val)
|
||||
if type_ == "unknown":
|
||||
continue
|
||||
indicators.append({"value": val, "type": type_})
|
||||
if len(indicators) >= cap:
|
||||
break
|
||||
|
||||
return {"source": cfg.get("feed_url"), "count": len(indicators), "indicators": indicators}
|
||||
|
||||
|
||||
_run(main)
|
||||
@@ -0,0 +1,69 @@
|
||||
import json, os, sys, re, ssl, urllib.request, urllib.error
|
||||
|
||||
|
||||
def _cfg():
|
||||
return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
|
||||
|
||||
|
||||
def _inputs():
|
||||
return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
|
||||
|
||||
|
||||
def _ctx(cfg):
|
||||
if cfg.get("insecure"):
|
||||
c = ssl.create_default_context()
|
||||
c.check_hostname = False
|
||||
c.verify_mode = ssl.CERT_NONE
|
||||
return c
|
||||
return None
|
||||
|
||||
|
||||
def _fetch(cfg):
|
||||
url = str(cfg.get("feed_url", ""))
|
||||
if not url:
|
||||
raise Exception("feed_url is not configured")
|
||||
headers = {"Accept": "text/plain, */*", "User-Agent": "Riposte-SOAR"}
|
||||
if cfg.get("api_token"):
|
||||
headers["Authorization"] = "Bearer " + str(cfg["api_token"])
|
||||
req = urllib.request.Request(url, headers=headers, method="GET")
|
||||
with urllib.request.urlopen(req, timeout=120, context=_ctx(cfg)) as r:
|
||||
return r.read().decode("utf-8", "replace")
|
||||
|
||||
|
||||
_IPV4 = re.compile(r"^\d{1,3}(?:\.\d{1,3}){3}(?:/\d{1,2})?$")
|
||||
_HASH = re.compile(r"^[a-fA-F0-9]{32}$|^[a-fA-F0-9]{40}$|^[a-fA-F0-9]{64}$")
|
||||
_EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||||
|
||||
|
||||
def detect_type(value):
|
||||
v = value.strip()
|
||||
if _IPV4.match(v) or (":" in v and re.match(r"^[0-9a-fA-F:]+$", v)):
|
||||
return "ip"
|
||||
if v.lower().startswith("http://") or v.lower().startswith("https://"):
|
||||
return "url"
|
||||
if _EMAIL.match(v):
|
||||
return "email"
|
||||
if _HASH.match(v):
|
||||
return "hash"
|
||||
if "." in v and " " not in v:
|
||||
return "domain"
|
||||
return "unknown"
|
||||
|
||||
|
||||
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):
|
||||
body = _fetch(cfg)
|
||||
return {"ok": True, "bytes": len(body)}
|
||||
|
||||
|
||||
_run(main)
|
||||
Reference in New Issue
Block a user