From f09d00114f8d85afbad73ce673522d3c2909d60e Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 22:43:16 +0200 Subject: [PATCH] feat(feed-csv): new generic CSV threat-intel feed connector Fetches a delimited feed URL and emits normalized IOCs {value,type} for TIM import (works with the /indicators/feed-extract -> /indicators/bulk flow). Column/delimiter/comment/type config, auto type-detection. New 'feed' category. Optional bearer auth, stdlib-only. py_compile clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/feed-csv/manifest.yaml | 49 ++++++++ .../feed-csv/scripts/fetch_indicators.py | 107 ++++++++++++++++++ .../feed-csv/scripts/test_connection.py | 69 +++++++++++ 3 files changed, 225 insertions(+) create mode 100644 integrations/feed-csv/manifest.yaml create mode 100644 integrations/feed-csv/scripts/fetch_indicators.py create mode 100644 integrations/feed-csv/scripts/test_connection.py diff --git a/integrations/feed-csv/manifest.yaml b/integrations/feed-csv/manifest.yaml new file mode 100644 index 0000000..a49cc54 --- /dev/null +++ b/integrations/feed-csv/manifest.yaml @@ -0,0 +1,49 @@ +id: feed_csv +name: CSV Feed +version: 1.0.0 +description: "Generic CSV threat-intel feed connector — fetch a CSV (or delimited) indicator list from a URL 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 delimited feed URL." +category: feed + +# Per-instance configuration. The feed is fetched over HTTP(S) from feed_url. +# An optional bearer token is sent if the feed requires authentication. +config_schema: + properties: + feed_url: + type: string + description: "URL of the CSV/delimited feed" + 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-csv-fetch-indicators + description: "Fetch and parse the CSV feed, returning normalized indicators." + risk: read + inputs_schema: + properties: + value_column: { type: number, description: "0-based column index holding the IOC value (default 0)" } + delimiter: { type: string, description: "Field delimiter (default ',')" } + ioc_type: { type: string, description: "auto | ip | domain | url | hash | email (default auto — detect per value)" } + skip_header: { type: boolean, description: "Skip the first row (default false)" } + comment_char: { type: string, description: "Lines starting with this are ignored (default '#')" } + max_indicators: { type: number, description: "Max indicators to return (default 10000)" } + required: [] + outputs_schema: { properties: {} } + + - id: test_connection + name: feed-csv-test-connection + description: "Verify the feed URL is reachable (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/feed-csv/scripts/fetch_indicators.py b/integrations/feed-csv/scripts/fetch_indicators.py new file mode 100644 index 0000000..8eb11ac --- /dev/null +++ b/integrations/feed-csv/scripts/fetch_indicators.py @@ -0,0 +1,107 @@ +import json, os, sys, io, csv, 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, text/csv, */*", "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) + + value_column = inputs.get("value_column", 0) + delimiter = inputs.get("delimiter", ",") + ioc_type = inputs.get("ioc_type", "auto") + skip_header = inputs.get("skip_header", False) + comment_char = inputs.get("comment_char", "#") + max_indicators = inputs.get("max_indicators", 10000) + + col = int(value_column or 0) + delim = (delimiter or ",") + if delim == "\\t": + delim = "\t" + cc = (comment_char or "#") + itype = (ioc_type or "auto") + cap = int(max_indicators or 10000) + + indicators = [] + header_skipped = False + reader = csv.reader(io.StringIO(text), delimiter=delim) + for row in reader: + if len(indicators) >= cap: + break + if not row: + continue + if row[0].strip().startswith(cc): + continue + if skip_header and not header_skipped: + header_skipped = True + continue + if len(row) > col: + val = row[col].strip() + if not val: + continue + t = itype if itype != "auto" else detect_type(val) + if t == "unknown": + continue + indicators.append({"value": val, "type": t}) + + return {"source": cfg.get("feed_url"), "count": len(indicators), "indicators": indicators} + + +_run(main) diff --git a/integrations/feed-csv/scripts/test_connection.py b/integrations/feed-csv/scripts/test_connection.py new file mode 100644 index 0000000..8da7434 --- /dev/null +++ b/integrations/feed-csv/scripts/test_connection.py @@ -0,0 +1,69 @@ +import json, os, sys, io, csv, 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, text/csv, */*", "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)