feat: Inital implementation of verify_certs

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):

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)

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)
 

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(

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"]