From 771c90ef0d8936c8e99bde880dcad7b2a2251d76 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 15:44:29 +0200 Subject: [PATCH] feat(imperva): new Imperva Cloud WAF integration Cloud Application Security API v1, 5 commands: list sites, get site status, block/whitelist IPs (site ACL). API-ID/key auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/imperva/manifest.yaml | 66 +++++++++++++++++++ integrations/imperva/scripts/block_ips.py | 52 +++++++++++++++ .../imperva/scripts/get_site_status.py | 45 +++++++++++++ integrations/imperva/scripts/list_sites.py | 42 ++++++++++++ .../imperva/scripts/test_connection.py | 43 ++++++++++++ integrations/imperva/scripts/whitelist_ips.py | 52 +++++++++++++++ 6 files changed, 300 insertions(+) create mode 100644 integrations/imperva/manifest.yaml create mode 100644 integrations/imperva/scripts/block_ips.py create mode 100644 integrations/imperva/scripts/get_site_status.py create mode 100644 integrations/imperva/scripts/list_sites.py create mode 100644 integrations/imperva/scripts/test_connection.py create mode 100644 integrations/imperva/scripts/whitelist_ips.py diff --git a/integrations/imperva/manifest.yaml b/integrations/imperva/manifest.yaml new file mode 100644 index 0000000..c3a67ec --- /dev/null +++ b/integrations/imperva/manifest.yaml @@ -0,0 +1,66 @@ +id: imperva +name: Imperva Cloud WAF +version: 1.0.0 +description: "Imperva Cloud WAF (Cloud Application Security API v1) — edge containment: list sites, read a site's status, block/unblock IPs via the site ACL, and set a security setting. API-ID/API-key authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: list sites, get site status, configure ACL (block IPs), configure security setting." +category: network + +# Per-instance configuration. api_id + api_key are sent as form parameters. +config_schema: + properties: + api_id: + type: string + description: "Imperva API ID" + api_key: + type: string + description: "Imperva API key" + x-soar-sensitive: true + required: + - api_id + - api_key + +commands: + - id: list_sites + name: imperva-list-sites + description: "List sites in the account." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + - id: get_site_status + name: imperva-get-site-status + description: "Get a site's status and configuration." + risk: read + inputs_schema: + properties: + site_id: { type: string, description: "Site ID" } + required: [site_id] + outputs_schema: { properties: {} } + - id: block_ips + name: imperva-block-ips + description: "Set the site's blacklisted-IPs ACL (replaces the current list)." + inputs_schema: + properties: + site_id: { type: string, description: "Site ID" } + ips: { type: string, description: "Comma-separated IPs/subnets to blacklist" } + required: [site_id, ips] + outputs_schema: { properties: {} } + - id: whitelist_ips + name: imperva-whitelist-ips + description: "Set the site's whitelisted-IPs ACL (replaces the current list)." + inputs_schema: + properties: + site_id: { type: string, description: "Site ID" } + ips: { type: string, description: "Comma-separated IPs/subnets to whitelist" } + required: [site_id, ips] + outputs_schema: { properties: {} } + + - id: test_connection + name: imperva-test-connection + description: "Verify the API credentials (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/imperva/scripts/block_ips.py b/integrations/imperva/scripts/block_ips.py new file mode 100644 index 0000000..70e3c3a --- /dev/null +++ b/integrations/imperva/scripts/block_ips.py @@ -0,0 +1,52 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://my.imperva.com/api/prov/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post(path, cfg, fields=None): + body = {"api_id": str(cfg.get("api_id", "")), "api_key": str(cfg.get("api_key", ""))} + if fields: + body.update({k: v for k, v in fields.items() if v not in (None, "")}) + data = urllib.parse.urlencode(body).encode("utf-8") + req = urllib.request.Request(BASE + path, data=data, + headers={"Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json"}, method="POST") + 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) + + +def main(cfg, inputs): + site_id = inputs.get("site_id") + if not site_id: + raise Exception("site_id is required") + ips = inputs.get("ips") + if not ips: + raise Exception("ips is required") + return post("/sites/configure/acl", cfg, { + "site_id": site_id, + "rule_id": "api.acl.blacklisted_ips", + "ips": ips, + }) + + +_run(main) diff --git a/integrations/imperva/scripts/get_site_status.py b/integrations/imperva/scripts/get_site_status.py new file mode 100644 index 0000000..24eac14 --- /dev/null +++ b/integrations/imperva/scripts/get_site_status.py @@ -0,0 +1,45 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://my.imperva.com/api/prov/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post(path, cfg, fields=None): + body = {"api_id": str(cfg.get("api_id", "")), "api_key": str(cfg.get("api_key", ""))} + if fields: + body.update({k: v for k, v in fields.items() if v not in (None, "")}) + data = urllib.parse.urlencode(body).encode("utf-8") + req = urllib.request.Request(BASE + path, data=data, + headers={"Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json"}, method="POST") + 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) + + +def main(cfg, inputs): + site_id = inputs.get("site_id") + if not site_id: + raise Exception("site_id is required") + return post("/sites/status", cfg, {"site_id": site_id}) + + +_run(main) diff --git a/integrations/imperva/scripts/list_sites.py b/integrations/imperva/scripts/list_sites.py new file mode 100644 index 0000000..ea727ce --- /dev/null +++ b/integrations/imperva/scripts/list_sites.py @@ -0,0 +1,42 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://my.imperva.com/api/prov/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post(path, cfg, fields=None): + body = {"api_id": str(cfg.get("api_id", "")), "api_key": str(cfg.get("api_key", ""))} + if fields: + body.update({k: v for k, v in fields.items() if v not in (None, "")}) + data = urllib.parse.urlencode(body).encode("utf-8") + req = urllib.request.Request(BASE + path, data=data, + headers={"Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json"}, method="POST") + 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) + + +def main(cfg, inputs): + return post("/sites/list", cfg) + + +_run(main) diff --git a/integrations/imperva/scripts/test_connection.py b/integrations/imperva/scripts/test_connection.py new file mode 100644 index 0000000..3e9ecb5 --- /dev/null +++ b/integrations/imperva/scripts/test_connection.py @@ -0,0 +1,43 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://my.imperva.com/api/prov/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post(path, cfg, fields=None): + body = {"api_id": str(cfg.get("api_id", "")), "api_key": str(cfg.get("api_key", ""))} + if fields: + body.update({k: v for k, v in fields.items() if v not in (None, "")}) + data = urllib.parse.urlencode(body).encode("utf-8") + req = urllib.request.Request(BASE + path, data=data, + headers={"Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json"}, method="POST") + 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) + + +def main(cfg, inputs): + post("/sites/list", cfg) + return {"ok": True} + + +_run(main) diff --git a/integrations/imperva/scripts/whitelist_ips.py b/integrations/imperva/scripts/whitelist_ips.py new file mode 100644 index 0000000..eb33bce --- /dev/null +++ b/integrations/imperva/scripts/whitelist_ips.py @@ -0,0 +1,52 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error + +BASE = "https://my.imperva.com/api/prov/v1" + + +def _cfg(): + return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}")) + + +def _inputs(): + return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}")) + + +def post(path, cfg, fields=None): + body = {"api_id": str(cfg.get("api_id", "")), "api_key": str(cfg.get("api_key", ""))} + if fields: + body.update({k: v for k, v in fields.items() if v not in (None, "")}) + data = urllib.parse.urlencode(body).encode("utf-8") + req = urllib.request.Request(BASE + path, data=data, + headers={"Content-Type": "application/x-www-form-urlencoded", + "Accept": "application/json"}, method="POST") + 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) + + +def main(cfg, inputs): + site_id = inputs.get("site_id") + if not site_id: + raise Exception("site_id is required") + ips = inputs.get("ips") + if not ips: + raise Exception("ips is required") + return post("/sites/configure/acl", cfg, { + "site_id": site_id, + "rule_id": "api.acl.whitelisted_ips", + "ips": ips, + }) + + +_run(main)