diff --git a/integrations/panos/manifest.yaml b/integrations/panos/manifest.yaml new file mode 100644 index 0000000..5e29027 --- /dev/null +++ b/integrations/panos/manifest.yaml @@ -0,0 +1,125 @@ +id: panos +name: PAN-OS / Panorama +version: 1.0.0 +description: "Palo Alto Networks PAN-OS / Panorama (XML API) — firewall containment: register/unregister IP tags for Dynamic Address Groups (fast block, no commit), create/delete address objects, create/edit custom URL categories, list and refresh External Dynamic Lists, and commit. API-key authentication; stdlib-only, no extra Python dependencies. Targets a firewall vsys." +changelog: "1.0.0 — Initial release: register/unregister IP tag, address create/delete, custom URL category create/edit, EDL list/refresh, commit." +category: network + +# Per-instance configuration. The API key is sent as the 'key' query parameter. +# vsys is the firewall virtual system (default vsys1). +config_schema: + properties: + server: + type: string + description: "PAN-OS / Panorama URL (e.g. https://192.168.0.1)" + api_key: + type: string + description: "PAN-OS API key" + x-soar-sensitive: true + port: + type: number + description: "HTTPS port (default 443)" + default: 443 + vsys: + type: string + description: "Virtual system (default vsys1)" + default: "vsys1" + insecure: + type: boolean + description: "Trust any TLS certificate (not secure)" + default: false + required: + - server + - api_key + +commands: + - id: register_ip_tag + name: panos-register-ip-tag + description: "Register (tag) an IP address so a Dynamic Address Group matching the tag blocks it immediately — no commit required." + inputs_schema: + properties: + ip: { type: string, description: "IP address to tag" } + tags: { type: string, description: "Comma-separated tags to attach" } + timeout: { type: number, description: "Tag timeout in seconds (0 or empty = no timeout)" } + required: [ip, tags] + outputs_schema: { properties: {} } + - id: unregister_ip_tag + name: panos-unregister-ip-tag + description: "Remove tags from an IP address (releases it from tag-based Dynamic Address Groups)." + inputs_schema: + properties: + ip: { type: string, description: "IP address to untag" } + tags: { type: string, description: "Comma-separated tags to remove" } + required: [ip, tags] + outputs_schema: { properties: {} } + - id: create_address + name: panos-create-address + description: "Create an address object (ip-netmask, ip-range, or fqdn)." + inputs_schema: + properties: + name: { type: string, description: "Address object name" } + value: { type: string, description: "Address value (e.g. 10.0.0.5/32, 10.0.0.1-10.0.0.9, or bad.example.com)" } + type: { type: string, description: "Address type: ip-netmask (default), ip-range, or fqdn" } + description: { type: string, description: "Optional description" } + required: [name, value] + outputs_schema: { properties: {} } + - id: delete_address + name: panos-delete-address + description: "Delete an address object by name." + inputs_schema: + properties: + name: { type: string, description: "Address object name" } + required: [name] + outputs_schema: { properties: {} } + - id: create_custom_url_category + name: panos-create-custom-url-category + description: "Create a custom URL category populated with the given URLs (used by a URL-filtering deny rule)." + inputs_schema: + properties: + name: { type: string, description: "Custom URL category name" } + urls: { type: string, description: "Comma-separated URLs/domains to include" } + description: { type: string, description: "Optional description" } + required: [name, urls] + outputs_schema: { properties: {} } + - id: edit_custom_url_category + name: panos-edit-custom-url-category + description: "Replace the URL list of an existing custom URL category." + inputs_schema: + properties: + name: { type: string, description: "Custom URL category name" } + urls: { type: string, description: "Comma-separated URLs/domains that become the category's full list" } + required: [name, urls] + outputs_schema: { properties: {} } + - id: list_edl + name: panos-list-edl + description: "List External Dynamic Lists configured on the device." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } + - id: refresh_edl + name: panos-refresh-edl + description: "Force an immediate refresh of an External Dynamic List." + inputs_schema: + properties: + name: { type: string, description: "External Dynamic List name" } + required: [name] + outputs_schema: { properties: {} } + - id: commit + name: panos-commit + description: "Commit the candidate configuration." + inputs_schema: + properties: + description: { type: string, description: "Optional commit description" } + required: [] + outputs_schema: { properties: {} } + + - id: test_connection + name: panos-test-connection + description: "Verify connectivity and the API key (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/panos/scripts/commit.py b/integrations/panos/scripts/commit.py new file mode 100644 index 0000000..982ada2 --- /dev/null +++ b/integrations/panos/scripts/commit.py @@ -0,0 +1,100 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET +from xml.sax.saxutils import escape + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + description = str(inputs.get("description") or "").strip() + + cmd = "" + if description: + cmd += "" + escape(description) + "" + cmd += "" + + root, text = api(cfg, {"type": "commit", "cmd": cmd}) + + job = root.findtext(".//job") + + return {"ok": True, "job": job} + + +_run(main) diff --git a/integrations/panos/scripts/create_address.py b/integrations/panos/scripts/create_address.py new file mode 100644 index 0000000..4503109 --- /dev/null +++ b/integrations/panos/scripts/create_address.py @@ -0,0 +1,111 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET +from xml.sax.saxutils import escape + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + name = str(inputs.get("name") or "").strip() + if not name: + raise Exception("name is required") + + value = str(inputs.get("value") or "").strip() + if not value: + raise Exception("value is required") + + addr_type = str(inputs.get("type") or "ip-netmask").strip() or "ip-netmask" + if addr_type not in ("ip-netmask", "ip-range", "fqdn"): + raise Exception("type must be one of ip-netmask, ip-range, fqdn") + + description = str(inputs.get("description") or "").strip() + + element = "<" + addr_type + ">" + escape(value) + "" + if description: + element += "" + escape(description) + "" + + xpath = _xpath_addr_prefix(cfg) + "/address/entry[@name='" + name + "']" + + api(cfg, {"type": "config", "action": "set", "xpath": xpath, "element": element}) + + return {"ok": True, "name": name} + + +_run(main) diff --git a/integrations/panos/scripts/create_custom_url_category.py b/integrations/panos/scripts/create_custom_url_category.py new file mode 100644 index 0000000..33ea08e --- /dev/null +++ b/integrations/panos/scripts/create_custom_url_category.py @@ -0,0 +1,110 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET +from xml.sax.saxutils import escape + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + name = str(inputs.get("name") or "").strip() + if not name: + raise Exception("name is required") + + urls = [s.strip() for s in str(inputs.get("urls") or "").split(",") if s.strip()] + if not urls: + raise Exception("urls is required") + + description = str(inputs.get("description") or "").strip() + + members = "".join("" + escape(u) + "" for u in urls) + element = "" + members + "URL List" + if description: + element += "" + escape(description) + "" + + xpath = ( + _xpath_addr_prefix(cfg) + "/profiles/custom-url-category/entry[@name='" + name + "']" + ) + + api(cfg, {"type": "config", "action": "set", "xpath": xpath, "element": element}) + + return {"ok": True, "name": name} + + +_run(main) diff --git a/integrations/panos/scripts/delete_address.py b/integrations/panos/scripts/delete_address.py new file mode 100644 index 0000000..a7e4965 --- /dev/null +++ b/integrations/panos/scripts/delete_address.py @@ -0,0 +1,96 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + name = str(inputs.get("name") or "").strip() + if not name: + raise Exception("name is required") + + xpath = _xpath_addr_prefix(cfg) + "/address/entry[@name='" + name + "']" + + api(cfg, {"type": "config", "action": "delete", "xpath": xpath}) + + return {"ok": True, "name": name} + + +_run(main) diff --git a/integrations/panos/scripts/edit_custom_url_category.py b/integrations/panos/scripts/edit_custom_url_category.py new file mode 100644 index 0000000..3699820 --- /dev/null +++ b/integrations/panos/scripts/edit_custom_url_category.py @@ -0,0 +1,107 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET +from xml.sax.saxutils import escape + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + name = str(inputs.get("name") or "").strip() + if not name: + raise Exception("name is required") + + urls = [s.strip() for s in str(inputs.get("urls") or "").split(",") if s.strip()] + if not urls: + raise Exception("urls is required") + + members = "".join("" + escape(u) + "" for u in urls) + element = "" + members + "" + + xpath = ( + _xpath_addr_prefix(cfg) + + "/profiles/custom-url-category/entry[@name='" + name + "']/list" + ) + + api(cfg, {"type": "config", "action": "edit", "xpath": xpath, "element": element}) + + return {"ok": True, "name": name} + + +_run(main) diff --git a/integrations/panos/scripts/list_edl.py b/integrations/panos/scripts/list_edl.py new file mode 100644 index 0000000..1c4485d --- /dev/null +++ b/integrations/panos/scripts/list_edl.py @@ -0,0 +1,96 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + xpath = _xpath_addr_prefix(cfg) + "/external-list" + + root, text = api(cfg, {"type": "config", "action": "get", "xpath": xpath}) + + result_elem = root.find("result") + if result_elem is None: + return {"result": {}} + + return {"result": _to_dict(result_elem)} + + +_run(main) diff --git a/integrations/panos/scripts/refresh_edl.py b/integrations/panos/scripts/refresh_edl.py new file mode 100644 index 0000000..c500159 --- /dev/null +++ b/integrations/panos/scripts/refresh_edl.py @@ -0,0 +1,101 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET +from xml.sax.saxutils import escape + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + name = str(inputs.get("name") or "").strip() + if not name: + raise Exception("name is required") + + cmd = ( + "" + + escape(name) + + "" + ) + + api(cfg, {"type": "op", "cmd": cmd}) + + return {"ok": True, "name": name} + + +_run(main) diff --git a/integrations/panos/scripts/register_ip_tag.py b/integrations/panos/scripts/register_ip_tag.py new file mode 100644 index 0000000..ddd5dff --- /dev/null +++ b/integrations/panos/scripts/register_ip_tag.py @@ -0,0 +1,119 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET +from xml.sax.saxutils import escape, quoteattr + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + ip = str(inputs.get("ip") or "").strip() + if not ip: + raise Exception("ip is required") + + tags = [s.strip() for s in str(inputs.get("tags") or "").split(",") if s.strip()] + if not tags: + raise Exception("tags is required") + + timeout_attr = "" + timeout = inputs.get("timeout") + if timeout not in (None, ""): + try: + t = int(timeout) + except (TypeError, ValueError): + t = 0 + if t > 0: + timeout_attr = ' timeout="' + str(t) + '"' + + members = "".join( + "" + escape(tag) + "" for tag in tags + ) + + uid_xml = ( + "1.0update" + "' + members + "" + "" + ) + + api(cfg, {"type": "user-id", "vsys": _vsys(cfg), "cmd": uid_xml}) + + return {"ok": True, "ip": ip, "tags": tags} + + +_run(main) diff --git a/integrations/panos/scripts/test_connection.py b/integrations/panos/scripts/test_connection.py new file mode 100644 index 0000000..4c2d5c3 --- /dev/null +++ b/integrations/panos/scripts/test_connection.py @@ -0,0 +1,92 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + cmd = "" + + api(cfg, {"type": "op", "cmd": cmd}) + + return {"ok": True} + + +_run(main) diff --git a/integrations/panos/scripts/unregister_ip_tag.py b/integrations/panos/scripts/unregister_ip_tag.py new file mode 100644 index 0000000..1c53677 --- /dev/null +++ b/integrations/panos/scripts/unregister_ip_tag.py @@ -0,0 +1,107 @@ +import json, os, sys, urllib.parse, urllib.request, urllib.error, ssl +import xml.etree.ElementTree as ET +from xml.sax.saxutils import escape, quoteattr + + +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 _base(cfg): + server = str(cfg.get("server", "")).rstrip("/") + port = cfg.get("port") or 443 + return server + ":" + str(port) + "/api/" + + +def _vsys(cfg): + return cfg.get("vsys") or "vsys1" + + +def _xpath_addr_prefix(cfg): + return ("/config/devices/entry[@name='localhost.localdomain']/vsys/entry[@name='" + + _vsys(cfg) + "']") + + +def api(cfg, params): + p = dict(params) + p["key"] = str(cfg.get("api_key", "")) + url = _base(cfg) + "?" + urllib.parse.urlencode(p) + req = urllib.request.Request(url, method="GET") + with urllib.request.urlopen(req, timeout=90, context=_ctx(cfg)) as r: + raw = r.read() + text = raw.decode("utf-8", "replace") + try: + root = ET.fromstring(raw) + except Exception: + raise Exception("Invalid XML response: " + text[:500]) + if root.attrib.get("status") != "success": + raise Exception("PAN-OS error: " + text[:800]) + return root, text + + +def _to_dict(elem): + # Shallow XML→dict for returning results + d = dict(elem.attrib) + children = list(elem) + if not children: + if elem.text and elem.text.strip(): + d["_text"] = elem.text.strip() + return d + for c in children: + cd = _to_dict(c) + if c.tag in d: + if not isinstance(d[c.tag], list): + d[c.tag] = [d[c.tag]] + d[c.tag].append(cd) + else: + d[c.tag] = cd + return d + + +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): + ip = str(inputs.get("ip") or "").strip() + if not ip: + raise Exception("ip is required") + + tags = [s.strip() for s in str(inputs.get("tags") or "").split(",") if s.strip()] + if not tags: + raise Exception("tags is required") + + members = "".join("" + escape(tag) + "" for tag in tags) + + uid_xml = ( + "1.0update" + "" + members + "" + "" + ) + + api(cfg, {"type": "user-id", "vsys": _vsys(cfg), "cmd": uid_xml}) + + return {"ok": True, "ip": ip, "tags": tags} + + +_run(main)