From 3a4d06441ce70224323fba4b567729b4d24a5b9e Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 16:07:37 +0200 Subject: [PATCH] feat(crowdsec): new CrowdSec CTI integration (French vendor) CrowdSec CTI API v2, 4 commands: IP reputation, multi-IP search, fire list. API-key auth, stdlib-only. py_compile clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/crowdsec/manifest.yaml | 54 +++++++++++++++++++ integrations/crowdsec/scripts/get_fire.py | 46 ++++++++++++++++ .../crowdsec/scripts/ip_reputation.py | 47 ++++++++++++++++ integrations/crowdsec/scripts/search_ips.py | 47 ++++++++++++++++ .../crowdsec/scripts/test_connection.py | 45 ++++++++++++++++ 5 files changed, 239 insertions(+) create mode 100644 integrations/crowdsec/manifest.yaml create mode 100644 integrations/crowdsec/scripts/get_fire.py create mode 100644 integrations/crowdsec/scripts/ip_reputation.py create mode 100644 integrations/crowdsec/scripts/search_ips.py create mode 100644 integrations/crowdsec/scripts/test_connection.py diff --git a/integrations/crowdsec/manifest.yaml b/integrations/crowdsec/manifest.yaml new file mode 100644 index 0000000..52940c8 --- /dev/null +++ b/integrations/crowdsec/manifest.yaml @@ -0,0 +1,54 @@ +id: crowdsec +name: CrowdSec CTI +version: 1.0.0 +description: "CrowdSec CTI (Cyber Threat Intelligence API v2) — IP reputation from CrowdSec's collaborative community: look up an IP's reputation and behaviors, search several IPs at once, and pull the currently-firing malicious IPs. API-key authentication; stdlib-only, no extra Python dependencies. (French vendor.)" +changelog: "1.0.0 — Initial release: IP reputation, multi-IP search, fire list." +category: enrichment + +# Per-instance configuration. The CTI API key is sent as the 'x-api-key' header. +config_schema: + properties: + api_key: + type: string + description: "CrowdSec CTI API key" + x-soar-sensitive: true + required: + - api_key + +commands: + - id: ip_reputation + name: crowdsec-ip-reputation + description: "Get an IP's CrowdSec reputation, behaviors and background noise score." + risk: read + inputs_schema: + properties: + ip: { type: string, description: "IP address" } + required: [ip] + outputs_schema: { properties: {} } + - id: search_ips + name: crowdsec-search-ips + description: "Look up several IPs in one request." + risk: read + inputs_schema: + properties: + ips: { type: string, description: "Comma-separated IP addresses" } + required: [ips] + outputs_schema: { properties: {} } + - id: get_fire + name: crowdsec-get-fire + description: "List IPs currently flagged as actively malicious (the fire list)." + risk: read + inputs_schema: + properties: + page: { type: number, description: "Page number (default 1)" } + required: [] + outputs_schema: { properties: {} } + + - id: test_connection + name: crowdsec-test-connection + description: "Verify the API key (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/crowdsec/scripts/get_fire.py b/integrations/crowdsec/scripts/get_fire.py new file mode 100644 index 0000000..4d7d0bf --- /dev/null +++ b/integrations/crowdsec/scripts/get_fire.py @@ -0,0 +1,46 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://cti.api.crowdsec.net/v2" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + url = BASE + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + req = urllib.request.Request(url, headers={"x-api-key": str(cfg.get("api_key", "")), "Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + page = inputs.get("page") + page = int(page) if page not in (None, "") else 1 + return request("/fire", cfg, params={"page": page}) + + +_run(main) diff --git a/integrations/crowdsec/scripts/ip_reputation.py b/integrations/crowdsec/scripts/ip_reputation.py new file mode 100644 index 0000000..4afc687 --- /dev/null +++ b/integrations/crowdsec/scripts/ip_reputation.py @@ -0,0 +1,47 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://cti.api.crowdsec.net/v2" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + url = BASE + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + req = urllib.request.Request(url, headers={"x-api-key": str(cfg.get("api_key", "")), "Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + ip = inputs.get("ip") + if not ip: + raise Exception("ip is required") + return request("/smoke/" + q(ip), cfg) + + +_run(main) diff --git a/integrations/crowdsec/scripts/search_ips.py b/integrations/crowdsec/scripts/search_ips.py new file mode 100644 index 0000000..2513645 --- /dev/null +++ b/integrations/crowdsec/scripts/search_ips.py @@ -0,0 +1,47 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://cti.api.crowdsec.net/v2" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + url = BASE + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + req = urllib.request.Request(url, headers={"x-api-key": str(cfg.get("api_key", "")), "Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + ips = inputs.get("ips") + if not ips: + raise Exception("ips is required") + return request("/smoke/search", cfg, params={"ips": ips}) + + +_run(main) diff --git a/integrations/crowdsec/scripts/test_connection.py b/integrations/crowdsec/scripts/test_connection.py new file mode 100644 index 0000000..4440ef6 --- /dev/null +++ b/integrations/crowdsec/scripts/test_connection.py @@ -0,0 +1,45 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://cti.api.crowdsec.net/v2" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def request(path, cfg, params=None): + url = BASE + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + req = urllib.request.Request(url, headers={"x-api-key": str(cfg.get("api_key", "")), "Accept": "application/json"}, method="GET") + 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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + request("/smoke/1.1.1.1", cfg) + return {"ok": True} + + +_run(main)