diff --git a/integrations/stormshield/manifest.yaml b/integrations/stormshield/manifest.yaml new file mode 100644 index 0000000..db9bb7a --- /dev/null +++ b/integrations/stormshield/manifest.yaml @@ -0,0 +1,75 @@ +id: stormshield +name: Stormshield Network Security +version: 1.0.0 +description: "Stormshield Network Security (SNS API) — firewall containment: authenticate, add or remove an IP from a block host-group object, and run a monitor query. Session-based authentication; stdlib-only, no extra Python dependencies. (French vendor.)" +changelog: "1.0.0 — Initial release: add/remove block-group member, list host objects, monitor query." +category: network + +# Per-instance configuration. Each command logs in (POST /api/auth/login) and +# reuses the returned session token for the request. +config_schema: + properties: + base_url: + type: string + description: "SNS appliance URL (e.g. https://sns.example.com)" + username: + type: string + description: "Admin username" + password: + type: string + description: "Admin password" + x-soar-sensitive: true + insecure: + type: boolean + description: "Trust any TLS certificate (not secure)" + default: false + required: + - base_url + - username + - password + +commands: + - id: add_block_group_member + name: stormshield-add-block-group-member + description: "Add an IP/host object to a block host-group (referenced by a filter rule)." + inputs_schema: + properties: + group: { type: string, description: "Host-group object name (e.g. RIPOSTE_BLOCK)" } + host: { type: string, description: "Host object name or IP to add" } + required: [group, host] + outputs_schema: { properties: {} } + - id: remove_block_group_member + name: stormshield-remove-block-group-member + description: "Remove a host from a block host-group." + inputs_schema: + properties: + group: { type: string, description: "Host-group object name" } + host: { type: string, description: "Host object name or IP to remove" } + required: [group, host] + outputs_schema: { properties: {} } + - id: list_hosts + name: stormshield-list-hosts + description: "List host objects." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + - id: monitor_query + name: stormshield-monitor-query + description: "Run a monitor command (e.g. MONITOR HOST) and return its result." + risk: read + inputs_schema: + properties: + command: { type: string, description: "SNS monitor command (e.g. 'MONITOR HOST')" } + required: [command] + outputs_schema: { properties: {} } + + - id: test_connection + name: stormshield-test-connection + description: "Verify connectivity and credentials by logging in (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/stormshield/scripts/add_block_group_member.py b/integrations/stormshield/scripts/add_block_group_member.py new file mode 100644 index 0000000..6840595 --- /dev/null +++ b/integrations/stormshield/scripts/add_block_group_member.py @@ -0,0 +1,96 @@ +import json, os, sys, ssl, base64, http.cookiejar +import urllib.parse, 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 _b64(s): + return base64.b64encode(str(s).encode("utf-8")).decode("ascii") + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({"uid": _b64(self.cfg.get("username", "")), "pswd": _b64(self.cfg.get("password", ""))}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/auth/login", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def command(self, cmd): + form = urllib.parse.urlencode({"cmd": cmd}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/command", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def logout(self): + try: + req = urllib.request.Request(self.base + "/api/auth/logout", data=b"", method="POST") + self.opener.open(req, timeout=30).read() + except Exception: + pass + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + try: + result = fn(client, inputs) + finally: + client.logout() + print(json.dumps(result)) + 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(client, inputs): + group = inputs.get("group") + host = inputs.get("host") + if not group: + raise Exception("group is required") + if not host: + raise Exception("host is required") + + add_result = client.command("CONFIG OBJECT GROUP ADDTO group=" + group + " node=" + host) + activate_result = client.command("CONFIG OBJECT ACTIVATE") + + return { + "group": group, + "host": host, + "add_result": add_result, + "activate_result": activate_result, + } + + +_run(main) diff --git a/integrations/stormshield/scripts/list_hosts.py b/integrations/stormshield/scripts/list_hosts.py new file mode 100644 index 0000000..aefff5f --- /dev/null +++ b/integrations/stormshield/scripts/list_hosts.py @@ -0,0 +1,82 @@ +import json, os, sys, ssl, base64, http.cookiejar +import urllib.parse, 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 _b64(s): + return base64.b64encode(str(s).encode("utf-8")).decode("ascii") + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({"uid": _b64(self.cfg.get("username", "")), "pswd": _b64(self.cfg.get("password", ""))}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/auth/login", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def command(self, cmd): + form = urllib.parse.urlencode({"cmd": cmd}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/command", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def logout(self): + try: + req = urllib.request.Request(self.base + "/api/auth/logout", data=b"", method="POST") + self.opener.open(req, timeout=30).read() + except Exception: + pass + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + try: + result = fn(client, inputs) + finally: + client.logout() + print(json.dumps(result)) + 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(client, inputs): + r = client.command("CONFIG OBJECT LIST type=host") + return {"result": r} + + +_run(main) diff --git a/integrations/stormshield/scripts/monitor_query.py b/integrations/stormshield/scripts/monitor_query.py new file mode 100644 index 0000000..2959321 --- /dev/null +++ b/integrations/stormshield/scripts/monitor_query.py @@ -0,0 +1,86 @@ +import json, os, sys, ssl, base64, http.cookiejar +import urllib.parse, 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 _b64(s): + return base64.b64encode(str(s).encode("utf-8")).decode("ascii") + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({"uid": _b64(self.cfg.get("username", "")), "pswd": _b64(self.cfg.get("password", ""))}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/auth/login", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def command(self, cmd): + form = urllib.parse.urlencode({"cmd": cmd}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/command", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def logout(self): + try: + req = urllib.request.Request(self.base + "/api/auth/logout", data=b"", method="POST") + self.opener.open(req, timeout=30).read() + except Exception: + pass + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + try: + result = fn(client, inputs) + finally: + client.logout() + print(json.dumps(result)) + 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(client, inputs): + command = inputs.get("command") + if not command: + raise Exception("command is required") + + r = client.command(command) + return {"command": command, "result": r} + + +_run(main) diff --git a/integrations/stormshield/scripts/remove_block_group_member.py b/integrations/stormshield/scripts/remove_block_group_member.py new file mode 100644 index 0000000..f0feca5 --- /dev/null +++ b/integrations/stormshield/scripts/remove_block_group_member.py @@ -0,0 +1,96 @@ +import json, os, sys, ssl, base64, http.cookiejar +import urllib.parse, 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 _b64(s): + return base64.b64encode(str(s).encode("utf-8")).decode("ascii") + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({"uid": _b64(self.cfg.get("username", "")), "pswd": _b64(self.cfg.get("password", ""))}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/auth/login", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def command(self, cmd): + form = urllib.parse.urlencode({"cmd": cmd}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/command", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def logout(self): + try: + req = urllib.request.Request(self.base + "/api/auth/logout", data=b"", method="POST") + self.opener.open(req, timeout=30).read() + except Exception: + pass + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + try: + result = fn(client, inputs) + finally: + client.logout() + print(json.dumps(result)) + 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(client, inputs): + group = inputs.get("group") + host = inputs.get("host") + if not group: + raise Exception("group is required") + if not host: + raise Exception("host is required") + + remove_result = client.command("CONFIG OBJECT GROUP REMOVEFROM group=" + group + " node=" + host) + activate_result = client.command("CONFIG OBJECT ACTIVATE") + + return { + "group": group, + "host": host, + "remove_result": remove_result, + "activate_result": activate_result, + } + + +_run(main) diff --git a/integrations/stormshield/scripts/test_connection.py b/integrations/stormshield/scripts/test_connection.py new file mode 100644 index 0000000..8d68a1a --- /dev/null +++ b/integrations/stormshield/scripts/test_connection.py @@ -0,0 +1,82 @@ +import json, os, sys, ssl, base64, http.cookiejar +import urllib.parse, 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 _b64(s): + return base64.b64encode(str(s).encode("utf-8")).decode("ascii") + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + ctx = _ctx(cfg) + self.opener = urllib.request.build_opener( + urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()), + urllib.request.HTTPSHandler(context=ctx) if ctx else urllib.request.HTTPSHandler(), + ) + + def login(self): + form = urllib.parse.urlencode({"uid": _b64(self.cfg.get("username", "")), "pswd": _b64(self.cfg.get("password", ""))}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/auth/login", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def command(self, cmd): + form = urllib.parse.urlencode({"cmd": cmd}).encode("utf-8") + req = urllib.request.Request(self.base + "/api/command", data=form, + headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST") + with self.opener.open(req, timeout=60) as r: + return r.read().decode("utf-8", "replace") + + def logout(self): + try: + req = urllib.request.Request(self.base + "/api/auth/logout", data=b"", method="POST") + self.opener.open(req, timeout=30).read() + except Exception: + pass + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.login() + try: + result = fn(client, inputs) + finally: + client.logout() + print(json.dumps(result)) + 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(client, inputs): + client.command("SYSTEM PROPERTY") + return {"ok": True} + + +_run(main)