From 6cc0bb61bca1ffef52b50fb1429e78122ca57eae Mon Sep 17 00:00:00 2001 From: Guillaume BOURGEOIS Date: Sun, 12 Jul 2026 00:20:41 +0200 Subject: [PATCH] feat(cisco-fmc): new Cisco Firepower Management Center integration FMC REST API, 7 commands: list/create/delete host objects, list access policies/rules, create block access rule. Token auth (generatetoken, token + domain UUID from response headers), stdlib-only. Co-Authored-By: Claude Opus 4.8 (1M context) --- integrations/cisco-fmc/manifest.yaml | 99 +++++++++++++++ .../cisco-fmc/scripts/create_access_rule.py | 115 ++++++++++++++++++ .../cisco-fmc/scripts/create_host_object.py | 90 ++++++++++++++ .../cisco-fmc/scripts/delete_host_object.py | 86 +++++++++++++ .../cisco-fmc/scripts/list_access_policies.py | 81 ++++++++++++ .../cisco-fmc/scripts/list_access_rules.py | 88 ++++++++++++++ .../cisco-fmc/scripts/list_host_objects.py | 81 ++++++++++++ .../cisco-fmc/scripts/test_connection.py | 81 ++++++++++++ 8 files changed, 721 insertions(+) create mode 100644 integrations/cisco-fmc/manifest.yaml create mode 100644 integrations/cisco-fmc/scripts/create_access_rule.py create mode 100644 integrations/cisco-fmc/scripts/create_host_object.py create mode 100644 integrations/cisco-fmc/scripts/delete_host_object.py create mode 100644 integrations/cisco-fmc/scripts/list_access_policies.py create mode 100644 integrations/cisco-fmc/scripts/list_access_rules.py create mode 100644 integrations/cisco-fmc/scripts/list_host_objects.py create mode 100644 integrations/cisco-fmc/scripts/test_connection.py diff --git a/integrations/cisco-fmc/manifest.yaml b/integrations/cisco-fmc/manifest.yaml new file mode 100644 index 0000000..fdced62 --- /dev/null +++ b/integrations/cisco-fmc/manifest.yaml @@ -0,0 +1,99 @@ +id: cisco_fmc +name: Cisco Firepower Management Center +version: 1.0.0 +description: "Cisco Secure Firewall Management Center (FMC REST API) — firewall containment: list and create/delete network host objects, list access policies and rules, and add a block access rule. Token authentication (generatetoken); stdlib-only, no extra Python dependencies." +changelog: "1.0.0 — Initial release: list/create/delete host objects, list access policies/rules, create block access rule." +category: network + +# Per-instance configuration. Auth calls generatetoken with HTTP Basic and reuses +# the returned X-auth-access-token and DOMAIN_UUID (default domain). +config_schema: + properties: + base_url: + type: string + description: "FMC base URL (e.g. https://fmc.example.com)" + username: + type: string + description: "FMC username" + password: + type: string + description: "FMC 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_host_objects + name: fmc-list-host-objects + description: "List network host objects." + risk: read + inputs_schema: + properties: + limit: { type: number, description: "Max objects (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: create_host_object + name: fmc-create-host-object + description: "Create a network host object." + inputs_schema: + properties: + name: { type: string, description: "Object name" } + value: { type: string, description: "Host IP address" } + description: { type: string, description: "Optional description" } + required: [name, value] + outputs_schema: { properties: {} } + - id: delete_host_object + name: fmc-delete-host-object + description: "Delete a network host object by ID." + inputs_schema: + properties: + object_id: { type: string, description: "Host object ID" } + required: [object_id] + outputs_schema: { properties: {} } + - id: list_access_policies + name: fmc-list-access-policies + description: "List access control policies." + risk: read + inputs_schema: + properties: + limit: { type: number, description: "Max policies (default 50)" } + required: [] + outputs_schema: { properties: {} } + - id: list_access_rules + name: fmc-list-access-rules + description: "List the access rules of an access control policy." + risk: read + inputs_schema: + properties: + policy_id: { type: string, description: "Access control policy ID" } + limit: { type: number, description: "Max rules (default 50)" } + required: [policy_id] + outputs_schema: { properties: {} } + - id: create_access_rule + name: fmc-create-access-rule + description: "Add an access rule to a policy (e.g. a BLOCK rule for source/destination network objects)." + inputs_schema: + properties: + policy_id: { type: string, description: "Access control policy ID" } + name: { type: string, description: "Rule name" } + action: { type: string, description: "ALLOW, TRUST, or BLOCK (default BLOCK)" } + source_object_ids: { type: string, description: "Comma-separated source network object IDs (optional)" } + destination_object_ids: { type: string, description: "Comma-separated destination network object IDs (optional)" } + enabled: { type: boolean, description: "Enable the rule (default true)" } + required: [policy_id, name] + outputs_schema: { properties: {} } + + - id: test_connection + name: fmc-test-connection + description: "Verify connectivity and credentials via generatetoken (used by the Test button)." + risk: read + inputs_schema: + properties: {} + required: [] + outputs_schema: { properties: {} } diff --git a/integrations/cisco-fmc/scripts/create_access_rule.py b/integrations/cisco-fmc/scripts/create_access_rule.py new file mode 100644 index 0000000..e9e4ba9 --- /dev/null +++ b/integrations/cisco-fmc/scripts/create_access_rule.py @@ -0,0 +1,115 @@ +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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + self.ctx = _ctx(cfg) + self.token = None + self.domain_uuid = None + + def authenticate(self): + creds = str(self.cfg.get("username", "")) + ":" + str(self.cfg.get("password", "")) + basic = base64.b64encode(creds.encode("utf-8")).decode("utf-8") + req = urllib.request.Request(self.base + "/api/fmc_platform/v1/auth/generatetoken", + data=b"", headers={"Authorization": "Basic " + basic}, method="POST") + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as r: + r.read() + self.token = r.headers.get("X-auth-access-token") + self.domain_uuid = r.headers.get("DOMAIN_UUID") + if not self.token or not self.domain_uuid: + raise Exception("Authentication failed: missing token or domain UUID") + + def _cfg_base(self): + return self.base + "/api/fmc_config/v1/domain/" + self.domain_uuid + + def call(self, method, path, body=None, params=None): + # path is relative to the domain config base, e.g. /object/hosts + url = self._cfg_base() + 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 = {"X-auth-access-token": self.token, "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=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.authenticate() + print(json.dumps(fn(client, 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 _csv(v): + return [s.strip() for s in str(v).split(",") if s.strip()] + + +def main(client, inputs): + policy_id = inputs.get("policy_id") + name = inputs.get("name") + action = inputs.get("action") + source_object_ids = inputs.get("source_object_ids") + destination_object_ids = inputs.get("destination_object_ids") + enabled = inputs.get("enabled") + if not policy_id: + raise Exception("policy_id is required") + if not name: + raise Exception("name is required") + + body = { + "name": name, + "action": (action or "BLOCK"), + "enabled": (enabled if enabled is not None else True), + "type": "AccessRule", + } + + if source_object_ids: + source_ids_list = _csv(source_object_ids) + if source_ids_list: + body["sourceNetworks"] = {"objects": [{"id": oid, "type": "Host"} for oid in source_ids_list]} + + if destination_object_ids: + dest_ids_list = _csv(destination_object_ids) + if dest_ids_list: + body["destinationNetworks"] = {"objects": [{"id": oid, "type": "Host"} for oid in dest_ids_list]} + + return client.call("POST", "/policy/accesspolicies/" + q(policy_id) + "/accessrules", body=body) + + +_run(main) diff --git a/integrations/cisco-fmc/scripts/create_host_object.py b/integrations/cisco-fmc/scripts/create_host_object.py new file mode 100644 index 0000000..0cfe98b --- /dev/null +++ b/integrations/cisco-fmc/scripts/create_host_object.py @@ -0,0 +1,90 @@ +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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + self.ctx = _ctx(cfg) + self.token = None + self.domain_uuid = None + + def authenticate(self): + creds = str(self.cfg.get("username", "")) + ":" + str(self.cfg.get("password", "")) + basic = base64.b64encode(creds.encode("utf-8")).decode("utf-8") + req = urllib.request.Request(self.base + "/api/fmc_platform/v1/auth/generatetoken", + data=b"", headers={"Authorization": "Basic " + basic}, method="POST") + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as r: + r.read() + self.token = r.headers.get("X-auth-access-token") + self.domain_uuid = r.headers.get("DOMAIN_UUID") + if not self.token or not self.domain_uuid: + raise Exception("Authentication failed: missing token or domain UUID") + + def _cfg_base(self): + return self.base + "/api/fmc_config/v1/domain/" + self.domain_uuid + + def call(self, method, path, body=None, params=None): + # path is relative to the domain config base, e.g. /object/hosts + url = self._cfg_base() + 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 = {"X-auth-access-token": self.token, "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=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.authenticate() + print(json.dumps(fn(client, 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(client, inputs): + name = inputs.get("name") + value = inputs.get("value") + description = inputs.get("description") + if not name: + raise Exception("name is required") + if not value: + raise Exception("value is required") + body = {"name": name, "type": "Host", "value": value} + if description: + body["description"] = description + return client.call("POST", "/object/hosts", body=body) + + +_run(main) diff --git a/integrations/cisco-fmc/scripts/delete_host_object.py b/integrations/cisco-fmc/scripts/delete_host_object.py new file mode 100644 index 0000000..e34a393 --- /dev/null +++ b/integrations/cisco-fmc/scripts/delete_host_object.py @@ -0,0 +1,86 @@ +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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + self.ctx = _ctx(cfg) + self.token = None + self.domain_uuid = None + + def authenticate(self): + creds = str(self.cfg.get("username", "")) + ":" + str(self.cfg.get("password", "")) + basic = base64.b64encode(creds.encode("utf-8")).decode("utf-8") + req = urllib.request.Request(self.base + "/api/fmc_platform/v1/auth/generatetoken", + data=b"", headers={"Authorization": "Basic " + basic}, method="POST") + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as r: + r.read() + self.token = r.headers.get("X-auth-access-token") + self.domain_uuid = r.headers.get("DOMAIN_UUID") + if not self.token or not self.domain_uuid: + raise Exception("Authentication failed: missing token or domain UUID") + + def _cfg_base(self): + return self.base + "/api/fmc_config/v1/domain/" + self.domain_uuid + + def call(self, method, path, body=None, params=None): + # path is relative to the domain config base, e.g. /object/hosts + url = self._cfg_base() + 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 = {"X-auth-access-token": self.token, "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=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.authenticate() + print(json.dumps(fn(client, 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(client, inputs): + object_id = inputs.get("object_id") + if not object_id: + raise Exception("object_id is required") + return client.call("DELETE", "/object/hosts/" + q(object_id)) + + +_run(main) diff --git a/integrations/cisco-fmc/scripts/list_access_policies.py b/integrations/cisco-fmc/scripts/list_access_policies.py new file mode 100644 index 0000000..3905064 --- /dev/null +++ b/integrations/cisco-fmc/scripts/list_access_policies.py @@ -0,0 +1,81 @@ +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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + self.ctx = _ctx(cfg) + self.token = None + self.domain_uuid = None + + def authenticate(self): + creds = str(self.cfg.get("username", "")) + ":" + str(self.cfg.get("password", "")) + basic = base64.b64encode(creds.encode("utf-8")).decode("utf-8") + req = urllib.request.Request(self.base + "/api/fmc_platform/v1/auth/generatetoken", + data=b"", headers={"Authorization": "Basic " + basic}, method="POST") + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as r: + r.read() + self.token = r.headers.get("X-auth-access-token") + self.domain_uuid = r.headers.get("DOMAIN_UUID") + if not self.token or not self.domain_uuid: + raise Exception("Authentication failed: missing token or domain UUID") + + def _cfg_base(self): + return self.base + "/api/fmc_config/v1/domain/" + self.domain_uuid + + def call(self, method, path, body=None, params=None): + # path is relative to the domain config base, e.g. /object/hosts + url = self._cfg_base() + 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 = {"X-auth-access-token": self.token, "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=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.authenticate() + print(json.dumps(fn(client, 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(client, inputs): + limit = inputs.get("limit") + return client.call("GET", "/policy/accesspolicies", params={"limit": int(limit or 50)}) + + +_run(main) diff --git a/integrations/cisco-fmc/scripts/list_access_rules.py b/integrations/cisco-fmc/scripts/list_access_rules.py new file mode 100644 index 0000000..2bfd7d8 --- /dev/null +++ b/integrations/cisco-fmc/scripts/list_access_rules.py @@ -0,0 +1,88 @@ +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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + self.ctx = _ctx(cfg) + self.token = None + self.domain_uuid = None + + def authenticate(self): + creds = str(self.cfg.get("username", "")) + ":" + str(self.cfg.get("password", "")) + basic = base64.b64encode(creds.encode("utf-8")).decode("utf-8") + req = urllib.request.Request(self.base + "/api/fmc_platform/v1/auth/generatetoken", + data=b"", headers={"Authorization": "Basic " + basic}, method="POST") + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as r: + r.read() + self.token = r.headers.get("X-auth-access-token") + self.domain_uuid = r.headers.get("DOMAIN_UUID") + if not self.token or not self.domain_uuid: + raise Exception("Authentication failed: missing token or domain UUID") + + def _cfg_base(self): + return self.base + "/api/fmc_config/v1/domain/" + self.domain_uuid + + def call(self, method, path, body=None, params=None): + # path is relative to the domain config base, e.g. /object/hosts + url = self._cfg_base() + 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 = {"X-auth-access-token": self.token, "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=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.authenticate() + print(json.dumps(fn(client, 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(client, inputs): + policy_id = inputs.get("policy_id") + limit = inputs.get("limit") + if not policy_id: + raise Exception("policy_id is required") + return client.call("GET", "/policy/accesspolicies/" + q(policy_id) + "/accessrules", + params={"limit": int(limit or 50), "expanded": "true"}) + + +_run(main) diff --git a/integrations/cisco-fmc/scripts/list_host_objects.py b/integrations/cisco-fmc/scripts/list_host_objects.py new file mode 100644 index 0000000..9b1a987 --- /dev/null +++ b/integrations/cisco-fmc/scripts/list_host_objects.py @@ -0,0 +1,81 @@ +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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + self.ctx = _ctx(cfg) + self.token = None + self.domain_uuid = None + + def authenticate(self): + creds = str(self.cfg.get("username", "")) + ":" + str(self.cfg.get("password", "")) + basic = base64.b64encode(creds.encode("utf-8")).decode("utf-8") + req = urllib.request.Request(self.base + "/api/fmc_platform/v1/auth/generatetoken", + data=b"", headers={"Authorization": "Basic " + basic}, method="POST") + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as r: + r.read() + self.token = r.headers.get("X-auth-access-token") + self.domain_uuid = r.headers.get("DOMAIN_UUID") + if not self.token or not self.domain_uuid: + raise Exception("Authentication failed: missing token or domain UUID") + + def _cfg_base(self): + return self.base + "/api/fmc_config/v1/domain/" + self.domain_uuid + + def call(self, method, path, body=None, params=None): + # path is relative to the domain config base, e.g. /object/hosts + url = self._cfg_base() + 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 = {"X-auth-access-token": self.token, "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=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.authenticate() + print(json.dumps(fn(client, 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(client, inputs): + limit = inputs.get("limit") + return client.call("GET", "/object/hosts", params={"limit": int(limit or 50), "expanded": "true"}) + + +_run(main) diff --git a/integrations/cisco-fmc/scripts/test_connection.py b/integrations/cisco-fmc/scripts/test_connection.py new file mode 100644 index 0000000..2939682 --- /dev/null +++ b/integrations/cisco-fmc/scripts/test_connection.py @@ -0,0 +1,81 @@ +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 + + +class Client: + def __init__(self, cfg): + self.cfg = cfg + self.base = str(cfg.get("base_url", "")).rstrip("/") + self.ctx = _ctx(cfg) + self.token = None + self.domain_uuid = None + + def authenticate(self): + creds = str(self.cfg.get("username", "")) + ":" + str(self.cfg.get("password", "")) + basic = base64.b64encode(creds.encode("utf-8")).decode("utf-8") + req = urllib.request.Request(self.base + "/api/fmc_platform/v1/auth/generatetoken", + data=b"", headers={"Authorization": "Basic " + basic}, method="POST") + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as r: + r.read() + self.token = r.headers.get("X-auth-access-token") + self.domain_uuid = r.headers.get("DOMAIN_UUID") + if not self.token or not self.domain_uuid: + raise Exception("Authentication failed: missing token or domain UUID") + + def _cfg_base(self): + return self.base + "/api/fmc_config/v1/domain/" + self.domain_uuid + + def call(self, method, path, body=None, params=None): + # path is relative to the domain config base, e.g. /object/hosts + url = self._cfg_base() + 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 = {"X-auth-access-token": self.token, "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=90, context=self.ctx) as r: + raw = r.read() + return json.loads(raw) if raw else {} + + +def _run(fn): + try: + cfg = _cfg() + inputs = _inputs() + client = Client(cfg) + client.authenticate() + print(json.dumps(fn(client, 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(client, inputs): + client.call("GET", "/object/hosts", params={"limit": 1}) + return {"ok": True, "domain_uuid": client.domain_uuid} + + +_run(main)