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) <noreply@anthropic.com>
2026-07-12 00:20:41 +02:00
parent 44d0836e1b
commit 496ac6d8ce
8 changed files with 719 additions and 0 deletions
@@ -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)