Guillaume BOURGEOIS
26cd02d77a
PAN-OS XML API, 10 commands: register/unregister IP tag (Dynamic Address Group fast-block, no commit), address create/delete, custom URL category create/edit, EDL list/refresh, commit. API-key auth, targets a vsys, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
112 lines
3.1 KiB
Python
112 lines
3.1 KiB
Python
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) + "</" + addr_type + ">"
|
|
if description:
|
|
element += "<description>" + escape(description) + "</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)
|