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) <noreply@anthropic.com>

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: {} }

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)

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)