From 9fd7de8b846e8dd3f6b27a0182d0b5421ba307df Mon Sep 17 00:00:00 2001 From: Johann Wagner Date: Wed, 29 Apr 2026 14:18:07 +0200 Subject: [PATCH] feat: Inital implementation of verify_certs --- cosmo/__main__.py | 6 +++++- cosmo/clients/netbox.py | 7 ++++++- cosmo/clients/netbox_client.py | 9 ++++++--- cosmo/clients/netbox_v4.py | 13 +++++++++++-- cosmo/config/cosmo_config.schema.json | 3 +++ 5 files changed, 31 insertions(+), 7 deletions(-) diff --git a/cosmo/__main__.py b/cosmo/__main__.py index 14d0f50d..1509b1bf 100644 --- a/cosmo/__main__.py +++ b/cosmo/__main__.py @@ -97,7 +97,11 @@ def main() -> int: if netbox_api_token is None: raise Exception("NETBOX_API_TOKEN is empty.") - nc = NetboxClient(url=netbox_url, token=netbox_api_token) + nc = NetboxClient( + url=netbox_url, + token=netbox_api_token, + verify_certs=cosmo_configuration.get("verify_certs", True), + ) cosmo_data = nc.get_data(cosmo_configuration["devices"]) def noop(*args, **kwargs): diff --git a/cosmo/clients/netbox.py b/cosmo/clients/netbox.py index aa3e0169..6571875a 100644 --- a/cosmo/clients/netbox.py +++ b/cosmo/clients/netbox.py @@ -10,9 +10,10 @@ class NetboxClient: - def __init__(self, url, token): + def __init__(self, url, token, verify_certs=True): self.url = url self.token = token + self.verify_certs = verify_certs version, feature_flags = self.query_version() base_version_match = re.search(r"[\d.]+", version) @@ -26,6 +27,7 @@ def __init__(self, url, token): multiple_mac_addresses=True, netbox_43_query_syntax=True, feature_flags=feature_flags, + verify_certs=self.verify_certs, ) elif self.base_version > Version("4.2.0"): log.info("Using version 4.2.x strategy...") @@ -35,6 +37,7 @@ def __init__(self, url, token): multiple_mac_addresses=True, netbox_43_query_syntax=False, feature_flags=feature_flags, + verify_certs=self.verify_certs, ) elif self.base_version > Version("4.0.0"): log.info("Using version 4.0.x strategy...") @@ -44,6 +47,7 @@ def __init__(self, url, token): multiple_mac_addresses=False, netbox_43_query_syntax=False, feature_flags=feature_flags, + verify_certs=self.verify_certs, ) else: raise Exception("Unknown Version") @@ -59,6 +63,7 @@ def query_version(self): "Content-Type": "application/json", "Accept": "application/json", }, + verify=self.verify_certs, ) if r.status_code != 200: raise Exception("Error querying api: " + r.text) diff --git a/cosmo/clients/netbox_client.py b/cosmo/clients/netbox_client.py index 07d35e2e..8856f900 100644 --- a/cosmo/clients/netbox_client.py +++ b/cosmo/clients/netbox_client.py @@ -8,10 +8,13 @@ class NetboxAPIClient: - def __init__(self, url, token, interprocess_shared_cache: DictProxy): + def __init__( + self, url, token, interprocess_shared_cache: DictProxy, verify_certs=True + ): self.url = url self.token = token self.cache = interprocess_shared_cache + self.verify_certs = verify_certs def query(self, query, query_name=None): @@ -25,6 +28,7 @@ def query(self, query, query_name=None): "Content-Type": "application/json", "Accept": "application/json", }, + verify=self.verify_certs, ) if r.status_code != 200: raise Exception("Error querying api: " + r.text) @@ -44,8 +48,7 @@ def query(self, query, query_name=None): def _cached_get(self, url, headers): if url not in self.cache: self.cache[url] = requests.get( - url, - headers=headers, + url, headers=headers, verify=self.verify_certs ) return self.cache.get(url) diff --git a/cosmo/clients/netbox_v4.py b/cosmo/clients/netbox_v4.py index 007ffe92..45b3d280 100644 --- a/cosmo/clients/netbox_v4.py +++ b/cosmo/clients/netbox_v4.py @@ -322,13 +322,20 @@ class NetboxV4Strategy: MAGIC_MAX_INFLIGHT = 15 def __init__( - self, url, token, multiple_mac_addresses, netbox_43_query_syntax, feature_flags + self, + url, + token, + multiple_mac_addresses, + netbox_43_query_syntax, + feature_flags, + verify_certs=True, ): self.url = url self.token = token self.multiple_mac_addresses = multiple_mac_addresses self.netbox_43_query_syntax = netbox_43_query_syntax self.feature_flags = feature_flags + self.verify_certs = verify_certs def worker_amount(self, n_queries: int): return clip(n_queries, self.MAGIC_MIN_INFLIGHT, self.MAGIC_MAX_INFLIGHT) @@ -345,7 +352,9 @@ def get_data(self, device_config): # https://stackoverflow.com/questions/72411392/can-you-do-nested-parallelization-using-multiprocessing-in-python # this avoids having to re-architecture completely using worker/task/queue model. with get_client_mp_context().Manager() as manager: - client = NetboxAPIClient(self.url, self.token, manager.dict()) + client = NetboxAPIClient( + self.url, self.token, manager.dict(), verify_certs=self.verify_certs + ) for d in device_list: queries.extend( diff --git a/cosmo/config/cosmo_config.schema.json b/cosmo/config/cosmo_config.schema.json index afdbccd8..9d0ccaff 100644 --- a/cosmo/config/cosmo_config.schema.json +++ b/cosmo/config/cosmo_config.schema.json @@ -23,6 +23,9 @@ "global_vrf": { "type": "string" }, + "verify_certs": { + "type": "boolean" + }, "output_format": { "type": "string", "enum": ["nix", "ansible"]