From 496ac6d8cec4ef100719a7788130c9050c350eef Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 00:20:41 +0200 Subject: [PATCH] feat(checkpoint): new Check Point firewall-containment integration Check Point Management Web API, 7 commands: show/add hosts, show access rulebase, add access rule (drop), publish, install policy. Session (login) auth with X-chkp-sid, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/checkpoint/manifest.yaml | 101 +++++++++++++++++ .../checkpoint/scripts/add_access_rule.py | 104 ++++++++++++++++++ integrations/checkpoint/scripts/add_host.py | 88 +++++++++++++++ .../checkpoint/scripts/install_policy.py | 92 ++++++++++++++++ integrations/checkpoint/scripts/publish.py | 82 ++++++++++++++ .../scripts/show_access_rulebase.py | 86 +++++++++++++++ integrations/checkpoint/scripts/show_hosts.py | 83 ++++++++++++++ .../checkpoint/scripts/test_connection.py | 83 ++++++++++++++ 8 files changed, 719 insertions(+) create mode 100644 integrations/checkpoint/manifest.yaml create mode 100644 integrations/checkpoint/scripts/add_access_rule.py create mode 100644 integrations/checkpoint/scripts/add_host.py create mode 100644 integrations/checkpoint/scripts/install_policy.py create mode 100644 integrations/checkpoint/scripts/publish.py create mode 100644 integrations/checkpoint/scripts/show_access_rulebase.py create mode 100644 integrations/checkpoint/scripts/show_hosts.py create mode 100644 integrations/checkpoint/scripts/test_connection.py diff --git a/integrations/checkpoint/manifest.yaml b/integrations/checkpoint/manifest.yaml new file mode 100644 index 0000000..d07a6f8 --- /dev/null +++ b/integrations/checkpoint/manifest.yaml @@ -0,0 +1,101 @@ +id: checkpoint +name: Check Point +version: 1.0.0 +description: "Check Point Management (Web API) — firewall containment: list and add host objects, view the access rulebase, add a drop rule (block), publish changes, and install policy. Session (login) authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: show/add hosts, show access rulebase, add access rule (drop), publish, install policy." +category: network + +# Per-instance configuration. Each command logs in to the Management API +# (returns a session id used as the X-chkp-sid header), performs the action, +# then logs out. Publishing/installing is explicit (separate commands). +config_schema: + properties: + server_url: + type: string + description: "Management server URL (e.g. https://mgmt.example.com)" + username: + type: string + description: "Management API username" + password: + type: string + description: "Management API password" + x-soar-sensitive: true + domain: + type: string + description: "Domain (for Multi-Domain Management; leave empty otherwise)" + insecure: + type: boolean + description: "Trust any TLS certificate (not secure)" + default: false + required: + - server_url + - username + - password + +commands: + - id: show_hosts + name: checkpoint-show-hosts + description: "List host objects." + risk: read + inputs_schema: + properties: + limit: { type: number, description: "Max hosts (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: add_host + name: checkpoint-add-host + description: "Create a host object." + inputs_schema: + properties: + name: { type: string, description: "Host object name" } + ip_address: { type: string, description: "Host IP address" } + required: [name, ip_address] + outputs_schema: { properties: {} } + - id: show_access_rulebase + name: checkpoint-show-access-rulebase + description: "Show the access rulebase of a policy layer." + risk: read + inputs_schema: + properties: + layer: { type: string, description: "Access layer name (e.g. Network)" } + limit: { type: number, description: "Max rules (default 50)" } + required: [layer] + outputs_schema: { properties: {} } + - id: add_access_rule + name: checkpoint-add-access-rule + description: "Add an access rule to a layer (e.g. a Drop rule to block a source)." + inputs_schema: + properties: + layer: { type: string, description: "Access layer name" } + name: { type: string, description: "Rule name" } + position: { type: string, description: "Position (e.g. top, bottom, or a number; default top)" } + source: { type: string, description: "Source object name (e.g. a host)" } + destination: { type: string, description: "Destination object name (default Any)" } + action: { type: string, description: "Accept, Drop, or Reject (default Drop)" } + required: [layer, name] + outputs_schema: { properties: {} } + - id: publish + name: checkpoint-publish + description: "Publish the current session's changes." + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + - id: install_policy + name: checkpoint-install-policy + description: "Install a policy package on gateway targets." + inputs_schema: + properties: + policy_package: { type: string, description: "Policy package name" } + targets: { type: string, description: "Comma-separated gateway target names" } + required: [policy_package, targets] + outputs_schema: { properties: {} } + + - id: test_connection + name: checkpoint-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/checkpoint/scripts/add_access_rule.py b/integrations/checkpoint/scripts/add_access_rule.py new file mode 100644 index 0000000..f8bec64 --- /dev/null +++ b/integrations/checkpoint/scripts/add_access_rule.py @@ -0,0 +1,104 @@ +import json, os, sys, 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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + "/web_api" + self.ctx = _ctx(cfg) + self.sid = None + + def _post(self, command, body, sid=None): + url = self.base + "/" + command + data = json.dumps(body if body is not None else {}).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if sid: + headers["X-chkp-sid"] = sid + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + def login(self): + body = {"user": self.cfg.get("username", ""), "password": self.cfg.get("password", "")} + if self.cfg.get("domain"): + body["domain"] = self.cfg["domain"] + resp = self._post("login", body) + self.sid = resp.get("sid") + if not self.sid: + raise Exception("Login failed: " + json.dumps(resp)) + return resp + + def call(self, command, body=None): + return self._post(command, body, sid=self.sid) + + def logout(self): + try: + self.call("logout", {}) + 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): + layer = inputs.get("layer") + if not layer: + raise Exception("layer is required") + name = inputs.get("name") + if not name: + raise Exception("name is required") + position = inputs.get("position") + action = inputs.get("action") + source = inputs.get("source") + destination = inputs.get("destination") + + body = { + "layer": layer, + "name": name, + "position": (position or "top"), + "action": (action or "Drop"), + } + if source: + body["source"] = source + if destination: + body["destination"] = destination + + return client.call("add-access-rule", body) + + +_run(main) diff --git a/integrations/checkpoint/scripts/add_host.py b/integrations/checkpoint/scripts/add_host.py new file mode 100644 index 0000000..788077f --- /dev/null +++ b/integrations/checkpoint/scripts/add_host.py @@ -0,0 +1,88 @@ +import json, os, sys, 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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + "/web_api" + self.ctx = _ctx(cfg) + self.sid = None + + def _post(self, command, body, sid=None): + url = self.base + "/" + command + data = json.dumps(body if body is not None else {}).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if sid: + headers["X-chkp-sid"] = sid + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + def login(self): + body = {"user": self.cfg.get("username", ""), "password": self.cfg.get("password", "")} + if self.cfg.get("domain"): + body["domain"] = self.cfg["domain"] + resp = self._post("login", body) + self.sid = resp.get("sid") + if not self.sid: + raise Exception("Login failed: " + json.dumps(resp)) + return resp + + def call(self, command, body=None): + return self._post(command, body, sid=self.sid) + + def logout(self): + try: + self.call("logout", {}) + 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): + name = inputs.get("name") + if not name: + raise Exception("name is required") + ip_address = inputs.get("ip_address") + if not ip_address: + raise Exception("ip_address is required") + return client.call("add-host", {"name": name, "ip-address": ip_address}) + + +_run(main) diff --git a/integrations/checkpoint/scripts/install_policy.py b/integrations/checkpoint/scripts/install_policy.py new file mode 100644 index 0000000..fd62af7 --- /dev/null +++ b/integrations/checkpoint/scripts/install_policy.py @@ -0,0 +1,92 @@ +import json, os, sys, 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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + "/web_api" + self.ctx = _ctx(cfg) + self.sid = None + + def _post(self, command, body, sid=None): + url = self.base + "/" + command + data = json.dumps(body if body is not None else {}).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if sid: + headers["X-chkp-sid"] = sid + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + def login(self): + body = {"user": self.cfg.get("username", ""), "password": self.cfg.get("password", "")} + if self.cfg.get("domain"): + body["domain"] = self.cfg["domain"] + resp = self._post("login", body) + self.sid = resp.get("sid") + if not self.sid: + raise Exception("Login failed: " + json.dumps(resp)) + return resp + + def call(self, command, body=None): + return self._post(command, body, sid=self.sid) + + def logout(self): + try: + self.call("logout", {}) + 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): + policy_package = inputs.get("policy_package") + if not policy_package: + raise Exception("policy_package is required") + targets = inputs.get("targets") + if not targets: + raise Exception("targets is required") + targets_list = [s.strip() for s in str(targets).split(",") if s.strip()] + if not targets_list: + raise Exception("targets is required") + + return client.call("install-policy", {"policy-package": policy_package, "targets": targets_list}) + + +_run(main) diff --git a/integrations/checkpoint/scripts/publish.py b/integrations/checkpoint/scripts/publish.py new file mode 100644 index 0000000..dce8eca --- /dev/null +++ b/integrations/checkpoint/scripts/publish.py @@ -0,0 +1,82 @@ +import json, os, sys, 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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + "/web_api" + self.ctx = _ctx(cfg) + self.sid = None + + def _post(self, command, body, sid=None): + url = self.base + "/" + command + data = json.dumps(body if body is not None else {}).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if sid: + headers["X-chkp-sid"] = sid + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + def login(self): + body = {"user": self.cfg.get("username", ""), "password": self.cfg.get("password", "")} + if self.cfg.get("domain"): + body["domain"] = self.cfg["domain"] + resp = self._post("login", body) + self.sid = resp.get("sid") + if not self.sid: + raise Exception("Login failed: " + json.dumps(resp)) + return resp + + def call(self, command, body=None): + return self._post(command, body, sid=self.sid) + + def logout(self): + try: + self.call("logout", {}) + 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): + return client.call("publish", {}) + + +_run(main) diff --git a/integrations/checkpoint/scripts/show_access_rulebase.py b/integrations/checkpoint/scripts/show_access_rulebase.py new file mode 100644 index 0000000..20f0093 --- /dev/null +++ b/integrations/checkpoint/scripts/show_access_rulebase.py @@ -0,0 +1,86 @@ +import json, os, sys, 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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + "/web_api" + self.ctx = _ctx(cfg) + self.sid = None + + def _post(self, command, body, sid=None): + url = self.base + "/" + command + data = json.dumps(body if body is not None else {}).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if sid: + headers["X-chkp-sid"] = sid + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + def login(self): + body = {"user": self.cfg.get("username", ""), "password": self.cfg.get("password", "")} + if self.cfg.get("domain"): + body["domain"] = self.cfg["domain"] + resp = self._post("login", body) + self.sid = resp.get("sid") + if not self.sid: + raise Exception("Login failed: " + json.dumps(resp)) + return resp + + def call(self, command, body=None): + return self._post(command, body, sid=self.sid) + + def logout(self): + try: + self.call("logout", {}) + 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): + layer = inputs.get("layer") + if not layer: + raise Exception("layer is required") + limit = inputs.get("limit") + return client.call("show-access-rulebase", {"name": layer, "limit": int(limit or 50)}) + + +_run(main) diff --git a/integrations/checkpoint/scripts/show_hosts.py b/integrations/checkpoint/scripts/show_hosts.py new file mode 100644 index 0000000..98624b2 --- /dev/null +++ b/integrations/checkpoint/scripts/show_hosts.py @@ -0,0 +1,83 @@ +import json, os, sys, 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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + "/web_api" + self.ctx = _ctx(cfg) + self.sid = None + + def _post(self, command, body, sid=None): + url = self.base + "/" + command + data = json.dumps(body if body is not None else {}).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if sid: + headers["X-chkp-sid"] = sid + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + def login(self): + body = {"user": self.cfg.get("username", ""), "password": self.cfg.get("password", "")} + if self.cfg.get("domain"): + body["domain"] = self.cfg["domain"] + resp = self._post("login", body) + self.sid = resp.get("sid") + if not self.sid: + raise Exception("Login failed: " + json.dumps(resp)) + return resp + + def call(self, command, body=None): + return self._post(command, body, sid=self.sid) + + def logout(self): + try: + self.call("logout", {}) + 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): + limit = inputs.get("limit") + return client.call("show-hosts", {"limit": int(limit or 50)}) + + +_run(main) diff --git a/integrations/checkpoint/scripts/test_connection.py b/integrations/checkpoint/scripts/test_connection.py new file mode 100644 index 0000000..3cb6903 --- /dev/null +++ b/integrations/checkpoint/scripts/test_connection.py @@ -0,0 +1,83 @@ +import json, os, sys, 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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("server_url", "")).rstrip("/") + "/web_api" + self.ctx = _ctx(cfg) + self.sid = None + + def _post(self, command, body, sid=None): + url = self.base + "/" + command + data = json.dumps(body if body is not None else {}).encode("utf-8") + headers = {"Content-Type": "application/json", "Accept": "application/json"} + if sid: + headers["X-chkp-sid"] = sid + req = urllib.request.Request(url, data=data, headers=headers, method="POST") + with urllib.request.urlopen(req, timeout=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + def login(self): + body = {"user": self.cfg.get("username", ""), "password": self.cfg.get("password", "")} + if self.cfg.get("domain"): + body["domain"] = self.cfg["domain"] + resp = self._post("login", body) + self.sid = resp.get("sid") + if not self.sid: + raise Exception("Login failed: " + json.dumps(resp)) + return resp + + def call(self, command, body=None): + return self._post(command, body, sid=self.sid) + + def logout(self): + try: + self.call("logout", {}) + 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.call("show-hosts", {"limit": 1}) + return {"ok": True} + + +_run(main)