Guillaume BOURGEOIS
496ac6d8ce
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>
93 lines
2.7 KiB
Python
93 lines
2.7 KiB
Python
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)
|