feat(cisco-ise): new Cisco ISE NAC integration

ISE ERS + ANC API, 6 commands: list/get endpoints, apply/clear ANC policy
(quarantine containment), list ANC policies. HTTP Basic auth, stdlib-only.
py_compile clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

integrations/automox/manifest.yaml

@@ -0,0 +1,66 @@
+id: automox
+name: Automox
+version: 1.0.0
+description: "Automox (API) — endpoint patch and configuration management: list and read devices, list policies, and queue a command (e.g. install updates or run a policy) on a device. API-key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: list/get devices, list policies, run device command."
+category: endpoint
+
+# Per-instance configuration. Auth header 'Authorization: Bearer <api_key>'.
+config_schema:
+  properties:
+    api_key:
+      type: string
+      description: "Automox API key"
+      x-soar-sensitive: true
+    org_id:
+      type: string
+      description: "Organization ID"
+  required:
+    - api_key
+    - org_id
+
+commands:
+  - id: list_devices
+    name: automox-list-devices
+    description: "List devices."
+    risk: read
+    inputs_schema:
+      properties:
+        limit: { type: number, description: "Max devices (default 50)" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: get_device
+    name: automox-get-device
+    description: "Get a single device by ID."
+    risk: read
+    inputs_schema:
+      properties:
+        device_id: { type: string, description: "Device (server) ID" }
+      required: [device_id]
+    outputs_schema: { properties: {} }
+  - id: list_policies
+    name: automox-list-policies
+    description: "List policies."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }
+  - id: run_command
+    name: automox-run-command
+    description: "Queue a command on a device (e.g. InstallUpdate, Reboot)."
+    inputs_schema:
+      properties:
+        device_id: { type: string, description: "Device (server) ID" }
+        command_type: { type: string, description: "Command type (e.g. InstallUpdate, Reboot, GetOS)" }
+      required: [device_id, command_type]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: automox-test-connection
+    description: "Verify the API key (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/automox/scripts/get_device.py

@@ -0,0 +1,49 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://console.automox.com/api"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(method, path, cfg, body=None, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["o"] = str(cfg.get("org_id", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "Bearer " + str(cfg.get("api_key", "")), "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) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    device_id = inputs.get("device_id")
+    if not device_id:
+        raise Exception("device_id is required")
+    return request("GET", "/servers/" + q(device_id), cfg)
+
+
+_run(main)

integrations/automox/scripts/list_devices.py

@@ -0,0 +1,48 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://console.automox.com/api"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(method, path, cfg, body=None, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["o"] = str(cfg.get("org_id", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "Bearer " + str(cfg.get("api_key", "")), "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) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    limit = inputs.get("limit")
+    limit = int(limit) if limit not in (None, "") else 50
+    return request("GET", "/servers", cfg, params={"limit": limit})
+
+
+_run(main)

integrations/automox/scripts/list_policies.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://console.automox.com/api"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(method, path, cfg, body=None, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["o"] = str(cfg.get("org_id", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "Bearer " + str(cfg.get("api_key", "")), "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) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    return request("GET", "/policies", cfg)
+
+
+_run(main)

integrations/automox/scripts/run_command.py

@@ -0,0 +1,55 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://console.automox.com/api"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(method, path, cfg, body=None, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["o"] = str(cfg.get("org_id", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "Bearer " + str(cfg.get("api_key", "")), "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) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    device_id = inputs.get("device_id")
+    if not device_id:
+        raise Exception("device_id is required")
+    command_type = inputs.get("command_type")
+    if not command_type:
+        raise Exception("command_type is required")
+    resp = request("POST", "/servers/" + q(device_id) + "/queues", cfg, body={"command_type_name": command_type})
+    if not resp:
+        return {"ok": True, "device_id": device_id, "command_type": command_type}
+    return resp
+
+
+_run(main)

integrations/automox/scripts/test_connection.py

@@ -0,0 +1,47 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://console.automox.com/api"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(method, path, cfg, body=None, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["o"] = str(cfg.get("org_id", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    data = json.dumps(body).encode("utf-8") if body is not None else None
+    headers = {"Authorization": "Bearer " + str(cfg.get("api_key", "")), "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) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    request("GET", "/servers", cfg, params={"limit": 1})
+    return {"ok": True}
+
+
+_run(main)

integrations/cisco-ise/manifest.yaml

@@ -0,0 +1,83 @@
+id: cisco_ise
+name: Cisco ISE
+version: 1.0.0
+description: "Cisco Identity Services Engine (ERS + ANC API) — network access containment: list and read endpoints, apply an Adaptive Network Control (ANC) quarantine policy to an endpoint, and clear it. HTTP Basic authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: list/get endpoints, apply/clear ANC policy, list ANC policies."
+category: network
+
+# Per-instance configuration. HTTP Basic auth against the ERS API (port 9060).
+config_schema:
+  properties:
+    base_url:
+      type: string
+      description: "ISE URL including the ERS port (e.g. https://ise.example.com:9060)"
+    username:
+      type: string
+      description: "ERS admin username"
+    password:
+      type: string
+      description: "ERS admin password"
+      x-soar-sensitive: true
+    insecure:
+      type: boolean
+      description: "Trust any TLS certificate (not secure)"
+      default: false
+  required:
+    - base_url
+    - username
+    - password
+
+commands:
+  - id: list_endpoints
+    name: ise-list-endpoints
+    description: "List endpoints (optionally filter by MAC)."
+    risk: read
+    inputs_schema:
+      properties:
+        mac: { type: string, description: "Optional MAC address filter" }
+      required: []
+    outputs_schema: { properties: {} }
+  - id: get_endpoint
+    name: ise-get-endpoint
+    description: "Get an endpoint by ID."
+    risk: read
+    inputs_schema:
+      properties:
+        endpoint_id: { type: string, description: "Endpoint ID" }
+      required: [endpoint_id]
+    outputs_schema: { properties: {} }
+  - id: apply_anc_policy
+    name: ise-apply-anc-policy
+    description: "Apply an ANC policy to an endpoint by MAC (quarantine — containment)."
+    inputs_schema:
+      properties:
+        mac: { type: string, description: "Endpoint MAC address" }
+        policy_name: { type: string, description: "ANC policy name (e.g. Quarantine)" }
+      required: [mac, policy_name]
+    outputs_schema: { properties: {} }
+  - id: clear_anc_policy
+    name: ise-clear-anc-policy
+    description: "Clear the ANC policy from an endpoint by MAC."
+    inputs_schema:
+      properties:
+        mac: { type: string, description: "Endpoint MAC address" }
+        policy_name: { type: string, description: "ANC policy name currently applied" }
+      required: [mac, policy_name]
+    outputs_schema: { properties: {} }
+  - id: list_anc_policies
+    name: ise-list-anc-policies
+    description: "List ANC policies."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: ise-test-connection
+    description: "Verify connectivity and credentials (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/cisco-ise/scripts/apply_anc_policy.py

@@ -0,0 +1,77 @@
+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 _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 = str(cfg.get("base_url", "")).rstrip("/") + "/ers/config" + 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()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    mac = inputs.get("mac")
+    policy_name = inputs.get("policy_name")
+    if not mac:
+        raise Exception("mac is required")
+    if not policy_name:
+        raise Exception("policy_name is required")
+    body = {
+        "OperationAdditionalData": {
+            "additionalData": [
+                {"name": "macAddress", "value": mac},
+                {"name": "policyName", "value": policy_name},
+            ]
+        }
+    }
+    result = request("PUT", "/ancendpoint/apply", cfg, body=body)
+    if not result:
+        return {"ok": True, "mac": mac, "policy": policy_name, "action": "apply"}
+    return result
+
+
+_run(main)

integrations/cisco-ise/scripts/clear_anc_policy.py

@@ -0,0 +1,77 @@
+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 _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 = str(cfg.get("base_url", "")).rstrip("/") + "/ers/config" + 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()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    mac = inputs.get("mac")
+    policy_name = inputs.get("policy_name")
+    if not mac:
+        raise Exception("mac is required")
+    if not policy_name:
+        raise Exception("policy_name is required")
+    body = {
+        "OperationAdditionalData": {
+            "additionalData": [
+                {"name": "macAddress", "value": mac},
+                {"name": "policyName", "value": policy_name},
+            ]
+        }
+    }
+    result = request("PUT", "/ancendpoint/clear", cfg, body=body)
+    if not result:
+        return {"ok": True, "mac": mac, "policy": policy_name, "action": "clear"}
+    return result
+
+
+_run(main)

integrations/cisco-ise/scripts/get_endpoint.py

@@ -0,0 +1,63 @@
+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 _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 = str(cfg.get("base_url", "")).rstrip("/") + "/ers/config" + 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()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    endpoint_id = inputs.get("endpoint_id")
+    if not endpoint_id:
+        raise Exception("endpoint_id is required")
+    return request("GET", "/endpoint/" + q(endpoint_id), cfg)
+
+
+_run(main)

integrations/cisco-ise/scripts/list_anc_policies.py

@@ -0,0 +1,60 @@
+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 _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 = str(cfg.get("base_url", "")).rstrip("/") + "/ers/config" + 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()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    return request("GET", "/ancpolicy", cfg)
+
+
+_run(main)

integrations/cisco-ise/scripts/list_endpoints.py

@@ -0,0 +1,62 @@
+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 _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 = str(cfg.get("base_url", "")).rstrip("/") + "/ers/config" + 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()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    mac = inputs.get("mac")
+    params = {"filter": "mac.EQ." + mac} if mac else {}
+    return request("GET", "/endpoint", cfg, params=params)
+
+
+_run(main)

integrations/cisco-ise/scripts/test_connection.py

@@ -0,0 +1,61 @@
+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 _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 = str(cfg.get("base_url", "")).rstrip("/") + "/ers/config" + 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()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    request("GET", "/endpoint", cfg, params={"size": 1})
+    return {"ok": True}
+
+
+_run(main)

integrations/domaintools/manifest.yaml

@@ -0,0 +1,67 @@
+id: domaintools
+name: DomainTools
+version: 1.0.0
+description: "DomainTools (Iris/Enterprise API) — domain and DNS intelligence: WHOIS lookup and history, domain reputation and risk scoring, and reverse-IP host domains. API-key authentication; stdlib-only, no extra Python dependencies."
+changelog: "1.0.0 — Initial release: whois, whois history, reputation, reverse IP."
+category: enrichment
+
+# Per-instance configuration. api_username + api_key are sent as query parameters.
+config_schema:
+  properties:
+    api_username:
+      type: string
+      description: "DomainTools API username"
+    api_key:
+      type: string
+      description: "DomainTools API key"
+      x-soar-sensitive: true
+  required:
+    - api_username
+    - api_key
+
+commands:
+  - id: whois
+    name: domaintools-whois
+    description: "Get current WHOIS for a domain."
+    risk: read
+    inputs_schema:
+      properties:
+        domain: { type: string, description: "Domain name" }
+      required: [domain]
+    outputs_schema: { properties: {} }
+  - id: whois_history
+    name: domaintools-whois-history
+    description: "Get historical WHOIS records for a domain."
+    risk: read
+    inputs_schema:
+      properties:
+        domain: { type: string, description: "Domain name" }
+      required: [domain]
+    outputs_schema: { properties: {} }
+  - id: reputation
+    name: domaintools-reputation
+    description: "Get a domain's reputation/risk score."
+    risk: read
+    inputs_schema:
+      properties:
+        domain: { type: string, description: "Domain name" }
+      required: [domain]
+    outputs_schema: { properties: {} }
+  - id: reverse_ip
+    name: domaintools-reverse-ip
+    description: "List domains hosted on an IP address."
+    risk: read
+    inputs_schema:
+      properties:
+        ip: { type: string, description: "IP address" }
+      required: [ip]
+    outputs_schema: { properties: {} }
+
+  - id: test_connection
+    name: domaintools-test-connection
+    description: "Verify the API credentials (used by the Test button)."
+    risk: read
+    inputs_schema:
+      properties: {}
+      required: []
+    outputs_schema: { properties: {} }

integrations/domaintools/scripts/reputation.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://api.domaintools.com"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(path, cfg, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["api_username"] = str(cfg.get("api_username", ""))
+    p["api_key"] = str(cfg.get("api_key", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    domain = inputs.get("domain")
+    if not domain:
+        raise Exception("domain is required")
+    return request("/v1/reputation", cfg, params={"domain": domain})
+
+
+_run(main)

integrations/domaintools/scripts/reverse_ip.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://api.domaintools.com"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(path, cfg, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["api_username"] = str(cfg.get("api_username", ""))
+    p["api_key"] = str(cfg.get("api_key", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    ip = inputs.get("ip")
+    if not ip:
+        raise Exception("ip is required")
+    return request("/v1/" + q(ip) + "/host-domains", cfg)
+
+
+_run(main)

integrations/domaintools/scripts/test_connection.py

@@ -0,0 +1,44 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://api.domaintools.com"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(path, cfg, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["api_username"] = str(cfg.get("api_username", ""))
+    p["api_key"] = str(cfg.get("api_key", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    request("/v1/account", cfg)
+    return {"ok": True}
+
+
+_run(main)

integrations/domaintools/scripts/whois.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://api.domaintools.com"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(path, cfg, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["api_username"] = str(cfg.get("api_username", ""))
+    p["api_key"] = str(cfg.get("api_key", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    domain = inputs.get("domain")
+    if not domain:
+        raise Exception("domain is required")
+    return request("/v1/" + q(domain) + "/whois", cfg)
+
+
+_run(main)

integrations/domaintools/scripts/whois_history.py

@@ -0,0 +1,46 @@
+import json, os, sys, urllib.parse, urllib.request, urllib.error
+
+BASE = "https://api.domaintools.com"
+
+
+def _cfg():
+    return json.loads(os.environ.get("INTEGRATION_SECRETS", "{}"))
+
+
+def _inputs():
+    return json.loads(os.environ.get("INTEGRATION_INPUTS", "{}"))
+
+
+def request(path, cfg, params=None):
+    p = {k: v for k, v in (params or {}).items() if v not in (None, "")}
+    p["api_username"] = str(cfg.get("api_username", ""))
+    p["api_key"] = str(cfg.get("api_key", ""))
+    url = BASE + path + "?" + urllib.parse.urlencode(p)
+    req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET")
+    with urllib.request.urlopen(req, timeout=60) as r:
+        raw = r.read()
+        return json.loads(raw) if raw else {}
+
+
+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):
+    domain = inputs.get("domain")
+    if not domain:
+        raise Exception("domain is required")
+    return request("/v1/" + q(domain) + "/whois/history", cfg)
+
+
+_run(main)