♻️ Moved sourcecode into netbox_zabbix_sync module

modules/config.py

@@ -1,134 +0,0 @@
-"""
-Module for parsing configuration from the top level config.py file
-"""
-
-from importlib import util
-from logging import getLogger
-from os import environ, path
-from pathlib import Path
-
-logger = getLogger(__name__)
-
-# PLEASE NOTE: This is a sample config file. Please do NOT make any edits in this file!
-# You should create your own config.py and it will overwrite the default config.
-
-DEFAULT_CONFIG = {
-    "templates_config_context": False,
-    "templates_config_context_overrule": False,
-    "template_cf": "zabbix_template",
-    "device_cf": "zabbix_hostid",
-    "proxy_cf": False,
-    "proxy_group_cf": False,
-    "clustering": False,
-    "create_hostgroups": True,
-    "create_journal": False,
-    "sync_vms": False,
-    "vm_hostgroup_format": "cluster_type/cluster/role",
-    "full_proxy_sync": False,
-    "zabbix_device_removal": ["Decommissioning", "Inventory"],
-    "zabbix_device_disable": ["Offline", "Planned", "Staged", "Failed"],
-    "hostgroup_format": "site/manufacturer/role",
-    "traverse_regions": False,
-    "traverse_site_groups": False,
-    "nb_device_filter": {"name__n": "null"},
-    "nb_vm_filter": {"name__n": "null"},
-    "inventory_mode": "disabled",
-    "inventory_sync": False,
-    "extended_site_properties": False,
-    "device_inventory_map": {
-        "asset_tag": "asset_tag",
-        "virtual_chassis/name": "chassis",
-        "status/label": "deployment_status",
-        "location/name": "location",
-        "latitude": "location_lat",
-        "longitude": "location_lon",
-        "comments": "notes",
-        "name": "name",
-        "rack/name": "site_rack",
-        "serial": "serialno_a",
-        "device_type/model": "type",
-        "device_type/manufacturer/name": "vendor",
-        "oob_ip/address": "oob_ip",
-    },
-    "vm_inventory_map": {
-        "status/label": "deployment_status",
-        "comments": "notes",
-        "name": "name",
-    },
-    "usermacro_sync": False,
-    "device_usermacro_map": {
-        "serial": "{$HW_SERIAL}",
-        "role/name": "{$DEV_ROLE}",
-        "url": "{$NB_URL}",
-        "id": "{$NB_ID}",
-    },
-    "vm_usermacro_map": {
-        "memory": "{$TOTAL_MEMORY}",
-        "role/name": "{$DEV_ROLE}",
-        "url": "{$NB_URL}",
-        "id": "{$NB_ID}",
-    },
-    "tag_sync": False,
-    "tag_lower": True,
-    "tag_name": "NetBox",
-    "tag_value": "name",
-    "device_tag_map": {
-        "site/name": "site",
-        "rack/name": "rack",
-        "platform/name": "target",
-    },
-    "vm_tag_map": {
-        "site/name": "site",
-        "cluster/name": "cluster",
-        "platform/name": "target",
-    },
-}
-
-
-def load_config():
-    """Returns combined config from all sources"""
-    # Overwrite default config with config.py
-    conf = load_config_file(config_default=DEFAULT_CONFIG)
-    # Overwrite default config and config.py with environment variables
-    for key in conf:
-        value_setting = load_env_variable(key)
-        if value_setting is not None:
-            conf[key] = value_setting
-    return conf
-
-
-def load_env_variable(config_environvar):
-    """Returns config from environment variable"""
-    prefix = "NBZX_"
-    config_environvar = prefix + config_environvar.upper()
-    if config_environvar in environ:
-        return environ[config_environvar]
-    return None
-
-
-def load_config_file(config_default, config_file="config.py"):
-    """Returns config from config.py file"""
-    # Find the script path and config file next to it.
-    script_dir = path.dirname(path.dirname(path.abspath(__file__)))
-    config_path = Path(path.join(script_dir, config_file))
-
-    # If the script directory is not found, try the current working directory
-    if not config_path.exists():
-        config_path = Path(config_file)
-
-    # If both checks fail then fallback to the default config
-    if not config_path.exists():
-        return config_default
-
-    dconf = config_default.copy()
-    # Dynamically import the config module
-    spec = util.spec_from_file_location("config", config_path)
-    if spec is None or spec.loader is None:
-        raise ImportError(f"Cannot load config from {config_path}")
-    config_module = util.module_from_spec(spec)
-    spec.loader.exec_module(config_module)
-    # Update DEFAULT_CONFIG with variables from the config module
-    for key in dconf:
-        if hasattr(config_module, key):
-            dconf[key] = getattr(config_module, key)
-    return dconf

netbox_zabbix_sync/modules/cli.py

@@ -0,0 +1,78 @@
+import argparse
+import logging
+from os import environ
+
+from netbox_zabbix_sync.modules.core import run_sync
+from netbox_zabbix_sync.modules.exceptions import EnvironmentVarError
+from netbox_zabbix_sync.modules.logging import get_logger, set_log_levels
+
+logger = get_logger()
+
+
+def main(arguments):
+    """Run the sync process."""
+    # set environment variables
+    if arguments.verbose:
+        set_log_levels(logging.WARNING, logging.INFO)
+    if arguments.debug:
+        set_log_levels(logging.WARNING, logging.DEBUG)
+    if arguments.debug_all:
+        set_log_levels(logging.DEBUG, logging.DEBUG)
+    if arguments.quiet:
+        set_log_levels(logging.ERROR, logging.ERROR)
+
+    # Gather environment variables for Zabbix and Netbox communication
+    env_vars = ["ZABBIX_HOST", "NETBOX_HOST", "NETBOX_TOKEN"]
+    if "ZABBIX_TOKEN" in environ:
+        env_vars.append("ZABBIX_TOKEN")
+    else:
+        env_vars.append("ZABBIX_USER")
+        env_vars.append("ZABBIX_PASS")
+    for var in env_vars:
+        if var not in environ:
+            e = f"Environment variable {var} has not been defined."
+            logger.error(e)
+            raise EnvironmentVarError(e)
+    # Get all virtual environment variables
+    if "ZABBIX_TOKEN" in env_vars:
+        zabbix_user = None
+        zabbix_pass = None
+        zabbix_token = environ.get("ZABBIX_TOKEN")
+    else:
+        zabbix_user = environ.get("ZABBIX_USER")
+        zabbix_pass = environ.get("ZABBIX_PASS")
+        zabbix_token = None
+    zabbix_host = environ.get("ZABBIX_HOST")
+    netbox_host = environ.get("NETBOX_HOST")
+    netbox_token = environ.get("NETBOX_TOKEN")
+
+    # Run main sync process
+    run_sync(
+        nb_host=netbox_host,
+        nb_token=netbox_token,
+        zbx_host=zabbix_host,
+        zbx_user=zabbix_user,
+        zbx_pass=zabbix_pass,
+        zbx_token=zabbix_token,
+    )
+
+
+def parse_cli():
+    parser = argparse.ArgumentParser(
+        description="A script to sync Zabbix with NetBox device data."
+    )
+    parser.add_argument(
+        "-v", "--verbose", help="Turn on verbose logging.", action="store_true"
+    )
+    parser.add_argument(
+        "-vv", "--debug", help="Turn on debugging.", action="store_true"
+    )
+    parser.add_argument(
+        "-vvv",
+        "--debug-all",
+        help="Turn on debugging for all modules.",
+        action="store_true",
+    )
+    parser.add_argument("-q", "--quiet", help="Turn off warnings.", action="store_true")
+    args = parser.parse_args()
+    main(args)

netbox_zabbix_sync/modules/core.py

@@ -1,5 +1,3 @@
-#!/usr/bin/env python3
-
 """Core component of the sync process"""
 
 import ssl
@@ -11,12 +9,16 @@ from pynetbox.core.query import RequestError as NBRequestError
 from requests.exceptions import ConnectionError as RequestsConnectionError
 from zabbix_utils import APIRequestError, ProcessingError, ZabbixAPI
 
-from modules.config import load_config
-from modules.device import PhysicalDevice
-from modules.exceptions import EnvironmentVarError, SyncError
-from modules.logging import get_logger, setup_logger
-from modules.tools import convert_recordset, proxy_prepper, verify_hg_format
-from modules.virtual_machine import VirtualMachine
+from netbox_zabbix_sync.modules.config import load_config
+from netbox_zabbix_sync.modules.device import PhysicalDevice
+from netbox_zabbix_sync.modules.exceptions import SyncError
+from netbox_zabbix_sync.modules.logging import get_logger, setup_logger
+from netbox_zabbix_sync.modules.tools import (
+    convert_recordset,
+    proxy_prepper,
+    verify_hg_format,
+)
+from netbox_zabbix_sync.modules.virtual_machine import VirtualMachine
 
 # Import configuration settings
 config = load_config()

netbox_zabbix_sync/modules/device.py

@@ -11,23 +11,23 @@ from typing import Any
 from pynetbox import RequestError as NetboxRequestError
 from zabbix_utils import APIRequestError
 
-from modules.config import load_config
-from modules.exceptions import (
+from netbox_zabbix_sync.modules.config import load_config
+from netbox_zabbix_sync.modules.exceptions import (
     InterfaceConfigError,
     SyncExternalError,
     SyncInventoryError,
     TemplateError,
 )
-from modules.hostgroups import Hostgroup
-from modules.interface import ZabbixInterface
-from modules.tags import ZabbixTags
-from modules.tools import (
+from netbox_zabbix_sync.modules.hostgroups import Hostgroup
+from netbox_zabbix_sync.modules.interface import ZabbixInterface
+from netbox_zabbix_sync.modules.tags import ZabbixTags
+from netbox_zabbix_sync.modules.tools import (
     cf_to_string,
     field_mapper,
     remove_duplicates,
     sanatize_log_output,
 )
-from modules.usermacros import ZabbixUsermacros
+from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros
 
 config = load_config()
 

netbox_zabbix_sync/modules/hostgroups.py

@@ -2,8 +2,8 @@
 
 from logging import getLogger
 
-from modules.exceptions import HostgroupError
-from modules.tools import build_path, cf_to_string
+from netbox_zabbix_sync.modules.exceptions import HostgroupError
+from netbox_zabbix_sync.modules.tools import build_path, cf_to_string
 
 
 class Hostgroup:

netbox_zabbix_sync/modules/interface.py

@@ -2,7 +2,7 @@
 All of the Zabbix interface related configuration
 """
 
-from modules.exceptions import InterfaceConfigError
+from netbox_zabbix_sync.modules.exceptions import InterfaceConfigError
 
 
 class ZabbixInterface:

netbox_zabbix_sync/modules/tags.py

@@ -4,7 +4,7 @@ All of the Zabbix Usermacro related configuration
 
 from logging import getLogger
 
-from modules.tools import field_mapper, remove_duplicates
+from netbox_zabbix_sync.modules.tools import field_mapper, remove_duplicates
 
 
 class ZabbixTags:

netbox_zabbix_sync/modules/tools.py

@@ -3,7 +3,7 @@
 from collections.abc import Callable
 from typing import Any, cast, overload
 
-from modules.exceptions import HostgroupError
+from netbox_zabbix_sync.modules.exceptions import HostgroupError
 
 
 def convert_recordset(recordset):

netbox_zabbix_sync/modules/usermacros.py

@@ -5,7 +5,7 @@ All of the Zabbix Usermacro related configuration
 from logging import getLogger
 from re import match
 
-from modules.tools import field_mapper, sanatize_log_output
+from netbox_zabbix_sync.modules.tools import field_mapper, sanatize_log_output
 
 
 class ZabbixUsermacros:

netbox_zabbix_sync/modules/virtual_machine.py

@@ -1,9 +1,13 @@
 """Module that hosts all functions for virtual machine processing"""
 
-from modules.config import load_config
-from modules.device import PhysicalDevice
-from modules.exceptions import InterfaceConfigError, SyncInventoryError, TemplateError
-from modules.interface import ZabbixInterface
+from netbox_zabbix_sync.modules.config import load_config
+from netbox_zabbix_sync.modules.device import PhysicalDevice
+from netbox_zabbix_sync.modules.exceptions import (
+    InterfaceConfigError,
+    SyncInventoryError,
+    TemplateError,
+)
+from netbox_zabbix_sync.modules.interface import ZabbixInterface
 
 # Load config
 config = load_config()

tests/test_configuration_parsing.py

@@ -3,7 +3,7 @@
 import os
 from unittest.mock import MagicMock, patch
 
-from modules.config import (
+from netbox_zabbix_sync.modules.config import (
     DEFAULT_CONFIG,
     load_config,
     load_config_file,

tests/test_device_deletion.py

@@ -5,8 +5,8 @@ from unittest.mock import MagicMock, patch
 
 from zabbix_utils import APIRequestError
 
-from modules.device import PhysicalDevice
-from modules.exceptions import SyncExternalError
+from netbox_zabbix_sync.modules.device import PhysicalDevice
+from netbox_zabbix_sync.modules.exceptions import SyncExternalError
 
 
 class TestDeviceDeletion(unittest.TestCase):

tests/test_hostgroups.py

@@ -3,8 +3,8 @@
 import unittest
 from unittest.mock import MagicMock, patch
 
-from modules.exceptions import HostgroupError
-from modules.hostgroups import Hostgroup
+from netbox_zabbix_sync.modules.exceptions import HostgroupError
+from netbox_zabbix_sync.modules.hostgroups import Hostgroup
 
 
 class TestHostgroups(unittest.TestCase):

tests/test_interface.py

@@ -3,8 +3,8 @@
 import unittest
 from typing import cast
 
-from modules.exceptions import InterfaceConfigError
-from modules.interface import ZabbixInterface
+from netbox_zabbix_sync.modules.exceptions import InterfaceConfigError
+from netbox_zabbix_sync.modules.interface import ZabbixInterface
 
 
 class TestZabbixInterface(unittest.TestCase):

tests/test_list_hostgroup_formats.py

@@ -3,9 +3,9 @@
 import unittest
 from unittest.mock import MagicMock
 
-from modules.exceptions import HostgroupError
-from modules.hostgroups import Hostgroup
-from modules.tools import verify_hg_format
+from netbox_zabbix_sync.modules.exceptions import HostgroupError
+from netbox_zabbix_sync.modules.hostgroups import Hostgroup
+from netbox_zabbix_sync.modules.tools import verify_hg_format
 
 
 class TestListHostgroupFormats(unittest.TestCase):

tests/test_physical_device.py

@@ -3,8 +3,8 @@
 import unittest
 from unittest.mock import MagicMock, patch
 
-from modules.device import PhysicalDevice
-from modules.exceptions import TemplateError
+from netbox_zabbix_sync.modules.device import PhysicalDevice
+from netbox_zabbix_sync.modules.exceptions import TemplateError
 
 
 class TestPhysicalDevice(unittest.TestCase):

tests/test_tools.py

@@ -1,4 +1,4 @@
-from modules.tools import sanatize_log_output
+from netbox_zabbix_sync.modules.tools import sanatize_log_output
 
 
 def test_sanatize_log_output_secrets():

tests/test_usermacros.py

@@ -1,8 +1,8 @@
 import unittest
 from unittest.mock import MagicMock, patch
 
-from modules.device import PhysicalDevice
-from modules.usermacros import ZabbixUsermacros
+from netbox_zabbix_sync.modules.device import PhysicalDevice
+from netbox_zabbix_sync.modules.usermacros import ZabbixUsermacros
 
 
 class DummyNB: