From d5b7e38dd732f3fe0a5d1ddc8e8813f6b2c3be38 Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 00:37:12 +0200 Subject: [PATCH] feat(infoblox): new Infoblox NIOS DNS-containment integration Infoblox WAPI, 7 commands: add RPZ domain (DNS sinkhole/block), list/delete RPZ records, search hosts, get networks, add A record. HTTP Basic auth, stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/infoblox/manifest.yaml | 101 ++++++++++++++++++ integrations/infoblox/scripts/add_a_record.py | 78 ++++++++++++++ .../infoblox/scripts/add_rpz_domain.py | 79 ++++++++++++++ .../infoblox/scripts/delete_rpz_record.py | 72 +++++++++++++ integrations/infoblox/scripts/get_network.py | 74 +++++++++++++ .../infoblox/scripts/list_rpz_records.py | 75 +++++++++++++ integrations/infoblox/scripts/search_host.py | 74 +++++++++++++ .../infoblox/scripts/test_connection.py | 65 +++++++++++ 8 files changed, 618 insertions(+) create mode 100644 integrations/infoblox/manifest.yaml create mode 100644 integrations/infoblox/scripts/add_a_record.py create mode 100644 integrations/infoblox/scripts/add_rpz_domain.py create mode 100644 integrations/infoblox/scripts/delete_rpz_record.py create mode 100644 integrations/infoblox/scripts/get_network.py create mode 100644 integrations/infoblox/scripts/list_rpz_records.py create mode 100644 integrations/infoblox/scripts/search_host.py create mode 100644 integrations/infoblox/scripts/test_connection.py diff --git a/integrations/infoblox/manifest.yaml b/integrations/infoblox/manifest.yaml new file mode 100644 index 0000000..999f159 --- /dev/null +++ b/integrations/infoblox/manifest.yaml @@ -0,0 +1,101 @@ +id: infoblox +name: Infoblox NIOS +version: 1.0.0 +description: "Infoblox NIOS (WAPI) — DNS-layer containment and DDI lookups: block a domain via an RPZ record (DNS sinkhole), list and delete RPZ records, search host records, look up networks, and add A records. HTTP Basic authentication; stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: add/list/delete RPZ records, search hosts, get networks, add A record." +category: network + +# Per-instance configuration. HTTP Basic auth against the Grid Master WAPI. +config_schema: + properties: + base_url: + type: string + description: "Grid Master URL (e.g. https://infoblox.example.com)" + username: + type: string + description: "WAPI username" + password: + type: string + description: "WAPI password" + x-soar-sensitive: true + wapi_version: + type: string + description: "WAPI version (default v2.12)" + default: "v2.12" + insecure: + type: boolean + description: "Trust any TLS certificate (not secure)" + default: false + required: + - base_url + - username + - password + +commands: + - id: add_rpz_domain + name: infoblox-add-rpz-domain + description: "Block a domain by adding an RPZ CNAME record (returns NXDOMAIN — DNS sinkhole)." + inputs_schema: + properties: + domain: { type: string, description: "Domain to block (e.g. bad.example.com)" } + rpz_zone: { type: string, description: "RPZ response-policy zone name (e.g. rpz.local)" } + view: { type: string, description: "DNS view (default 'default')" } + required: [domain, rpz_zone] + outputs_schema: { properties: {} } + - id: list_rpz_records + name: infoblox-list-rpz-records + description: "List RPZ CNAME records (optionally in a specific zone)." + risk: read + inputs_schema: + properties: + rpz_zone: { type: string, description: "Optional RPZ zone filter" } + max_results: { type: number, description: "Max records (default 100)" } + required: [] + outputs_schema: { properties: {} } + - id: delete_rpz_record + name: infoblox-delete-rpz-record + description: "Delete an RPZ record (unblock) by its object reference." + inputs_schema: + properties: + ref: { type: string, description: "WAPI object reference (_ref) of the record" } + required: [ref] + outputs_schema: { properties: {} } + - id: search_host + name: infoblox-search-host + description: "Search host records by name (substring match)." + risk: read + inputs_schema: + properties: + name: { type: string, description: "Host name (or partial)" } + max_results: { type: number, description: "Max records (default 100)" } + required: [name] + outputs_schema: { properties: {} } + - id: get_network + name: infoblox-get-network + description: "Look up networks (optionally filtered by CIDR)." + risk: read + inputs_schema: + properties: + network: { type: string, description: "Optional network CIDR filter (e.g. 10.0.0.0/24)" } + max_results: { type: number, description: "Max networks (default 100)" } + required: [] + outputs_schema: { properties: {} } + - id: add_a_record + name: infoblox-add-a-record + description: "Add a DNS A record." + inputs_schema: + properties: + name: { type: string, description: "FQDN of the record" } + ipv4addr: { type: string, description: "IPv4 address" } + view: { type: string, description: "DNS view (default 'default')" } + required: [name, ipv4addr] + outputs_schema: { properties: {} } + + - id: test_connection + name: infoblox-test-connection + description: "Verify connectivity and credentials (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/infoblox/scripts/add_a_record.py b/integrations/infoblox/scripts/add_a_record.py new file mode 100644 index 0000000..fda3a29 --- /dev/null +++ b/integrations/infoblox/scripts/add_a_record.py @@ -0,0 +1,78 @@ +import json, os, sys, base64, ssl, urllib.parse, 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 + + +def _base(cfg): + return str(cfg.get("base_url", "")).rstrip("/") + "/wapi/" + str(cfg.get("wapi_version") or "v2.12") + + +def _auth(cfg): + raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", "")) + return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8") + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"result": raw.decode("utf-8", "replace")} + + +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 = inputs.get("name") + ipv4addr = inputs.get("ipv4addr") + if not name: + raise Exception("name is required") + if not ipv4addr: + raise Exception("ipv4addr is required") + view = inputs.get("view") or "default" + + body = { + "name": name, + "ipv4addr": ipv4addr, + "view": view, + } + resp = request("POST", "/record:a", cfg, body=body) + return {"result": resp} + + +_run(main) diff --git a/integrations/infoblox/scripts/add_rpz_domain.py b/integrations/infoblox/scripts/add_rpz_domain.py new file mode 100644 index 0000000..9677170 --- /dev/null +++ b/integrations/infoblox/scripts/add_rpz_domain.py @@ -0,0 +1,79 @@ +import json, os, sys, base64, ssl, urllib.parse, 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 + + +def _base(cfg): + return str(cfg.get("base_url", "")).rstrip("/") + "/wapi/" + str(cfg.get("wapi_version") or "v2.12") + + +def _auth(cfg): + raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", "")) + return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8") + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"result": raw.decode("utf-8", "replace")} + + +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): + domain = inputs.get("domain") + rpz_zone = inputs.get("rpz_zone") + if not domain: + raise Exception("domain is required") + if not rpz_zone: + raise Exception("rpz_zone is required") + view = inputs.get("view") or "default" + + body = { + "name": str(domain) + "." + str(rpz_zone), + "canonical": "", + "rp_zone": rpz_zone, + "view": view, + } + resp = request("POST", "/record:rpz:cname", cfg, body=body) + return {"result": resp} + + +_run(main) diff --git a/integrations/infoblox/scripts/delete_rpz_record.py b/integrations/infoblox/scripts/delete_rpz_record.py new file mode 100644 index 0000000..08a362c --- /dev/null +++ b/integrations/infoblox/scripts/delete_rpz_record.py @@ -0,0 +1,72 @@ +import json, os, sys, base64, ssl, urllib.parse, 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 + + +def _base(cfg): + return str(cfg.get("base_url", "")).rstrip("/") + "/wapi/" + str(cfg.get("wapi_version") or "v2.12") + + +def _auth(cfg): + raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", "")) + return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8") + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"result": raw.decode("utf-8", "replace")} + + +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) + + +q = lambda v: urllib.parse.quote(str(v), safe="") + + +def main(cfg, inputs): + ref = inputs.get("ref") + if not ref: + raise Exception("ref is required") + + resp = request("DELETE", "/" + q(ref), cfg) + return {"deleted": resp} + + +_run(main) diff --git a/integrations/infoblox/scripts/get_network.py b/integrations/infoblox/scripts/get_network.py new file mode 100644 index 0000000..58844bb --- /dev/null +++ b/integrations/infoblox/scripts/get_network.py @@ -0,0 +1,74 @@ +import json, os, sys, base64, ssl, urllib.parse, 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 + + +def _base(cfg): + return str(cfg.get("base_url", "")).rstrip("/") + "/wapi/" + str(cfg.get("wapi_version") or "v2.12") + + +def _auth(cfg): + raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", "")) + return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8") + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"result": raw.decode("utf-8", "replace")} + + +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): + network = inputs.get("network") + max_results = inputs.get("max_results") + + params = { + "_max_results": int(max_results or 100), + } + if network: + params["network"] = network + + resp = request("GET", "/network", cfg, params=params) + return {"networks": resp} + + +_run(main) diff --git a/integrations/infoblox/scripts/list_rpz_records.py b/integrations/infoblox/scripts/list_rpz_records.py new file mode 100644 index 0000000..77e45be --- /dev/null +++ b/integrations/infoblox/scripts/list_rpz_records.py @@ -0,0 +1,75 @@ +import json, os, sys, base64, ssl, urllib.parse, 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 + + +def _base(cfg): + return str(cfg.get("base_url", "")).rstrip("/") + "/wapi/" + str(cfg.get("wapi_version") or "v2.12") + + +def _auth(cfg): + raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", "")) + return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8") + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"result": raw.decode("utf-8", "replace")} + + +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): + rpz_zone = inputs.get("rpz_zone") + max_results = inputs.get("max_results") + + params = { + "_max_results": int(max_results or 100), + "_return_fields+": "name,canonical,rp_zone,view", + } + if rpz_zone: + params["rp_zone"] = rpz_zone + + resp = request("GET", "/record:rpz:cname", cfg, params=params) + return {"records": resp} + + +_run(main) diff --git a/integrations/infoblox/scripts/search_host.py b/integrations/infoblox/scripts/search_host.py new file mode 100644 index 0000000..38d8caf --- /dev/null +++ b/integrations/infoblox/scripts/search_host.py @@ -0,0 +1,74 @@ +import json, os, sys, base64, ssl, urllib.parse, 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 + + +def _base(cfg): + return str(cfg.get("base_url", "")).rstrip("/") + "/wapi/" + str(cfg.get("wapi_version") or "v2.12") + + +def _auth(cfg): + raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", "")) + return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8") + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"result": raw.decode("utf-8", "replace")} + + +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 = inputs.get("name") + if not name: + raise Exception("name is required") + max_results = inputs.get("max_results") + + params = { + "name~": name, + "_max_results": int(max_results or 100), + } + resp = request("GET", "/record:host", cfg, params=params) + return {"records": resp} + + +_run(main) diff --git a/integrations/infoblox/scripts/test_connection.py b/integrations/infoblox/scripts/test_connection.py new file mode 100644 index 0000000..2094ec2 --- /dev/null +++ b/integrations/infoblox/scripts/test_connection.py @@ -0,0 +1,65 @@ +import json, os, sys, base64, ssl, urllib.parse, 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 + + +def _base(cfg): + return str(cfg.get("base_url", "")).rstrip("/") + "/wapi/" + str(cfg.get("wapi_version") or "v2.12") + + +def _auth(cfg): + raw = str(cfg.get("username", "")) + ":" + str(cfg.get("password", "")) + return "Basic " + base64.b64encode(raw.encode("utf-8")).decode("utf-8") + + +def request(method, path, cfg, body=None, params=None): + url = _base(cfg) + path + if params: + clean = {k: v for k, v in params.items() if v not in (None, "")} + if clean: + url += "?" + urllib.parse.urlencode(clean) + data = json.dumps(body).encode("utf-8") if body is not None else None + headers = {"Authorization": _auth(cfg), "Accept": "application/json"} + if data is not None: + headers["Content-Type"] = "application/json" + req = urllib.request.Request(url, data=data, headers=headers, method=method) + with urllib.request.urlopen(req, timeout=60, context=_ctx(cfg)) as r: + raw = r.read() + try: + return json.loads(raw) if raw else {} + except Exception: + return {"result": raw.decode("utf-8", "replace")} + + +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): + request("GET", "/grid", cfg) + return {"ok": True} + + +_run(main)