NeoAnd/netbox-zabbix-sync
1
0
Fork 0
mirror of https://github.com/TheNetworkGuy/netbox-zabbix-sync.git synced 2025-07-14 01:41:25 -06:00
Code Issues Actions Packages Projects Releases Wiki Activity

✨ Added support for custom CA contexts within ZabbixAPI

Browse Source
This commit is contained in:
Wouter de Bruijn 2024-12-05 13:59:12 +01:00
parent 4ec8036c88
commit 8b670ba395
No known key found for this signature in database
GPG Key ID: AC71F96733B92BFA
1 changed files with 32 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
netbox_zabbix_sync.py
View File

@ -4,6 +4,7 @@
"""Netbox to Zabbix sync script.""" """Netbox to Zabbix sync script."""
import logging import logging
import argparse import argparse
import ssl
from os import environ, path, sys from os import environ, path, sys
from pynetbox import api from pynetbox import api
from pynetbox.core.query import RequestError as NBRequestError from pynetbox.core.query import RequestError as NBRequestError
@ -49,6 +50,7 @@ logger.addHandler(lgout)
logger.addHandler(lgfile) logger.addHandler(lgfile)
logger.setLevel(logging.WARNING) logger.setLevel(logging.WARNING)
def main(arguments): def main(arguments):
"""Run the sync process.""" """Run the sync process."""
# pylint: disable=too-many-branches, too-many-statements # pylint: disable=too-many-branches, too-many-statements
@ -86,7 +88,8 @@ def main(arguments):
"site", "site_group", "tenant", "tenant_group"] "site", "site_group", "tenant", "tenant_group"]
# Create API call to get all custom fields which are on the device objects # Create API call to get all custom fields which are on the device objects
try: try:
device_cfs = list(netbox.extras.custom_fields.filter(type="text", content_type_id=23)) device_cfs = list(netbox.extras.custom_fields.filter(
type="text", content_type_id=23))
except RequestsConnectionError: except RequestsConnectionError:
logger.error(f"Unable to connect to Netbox with URL {netbox_host}." logger.error(f"Unable to connect to Netbox with URL {netbox_host}."
" Please check the URL and status of Netbox.") " Please check the URL and status of Netbox.")
@ -104,10 +107,18 @@ def main(arguments):
raise HostgroupError(e) raise HostgroupError(e)
# Set Zabbix API # Set Zabbix API
try: try:
ssl_ctx = ssl.create_default_context()
# If a custom CA bundle is set for pynetbox (requests), also use it for the Zabbix API
if environ.get("REQUESTS_CA_BUNDLE", None):
ssl_ctx.load_verify_locations(environ["REQUESTS_CA_BUNDLE"])
if not zabbix_token: if not zabbix_token:
zabbix = ZabbixAPI(zabbix_host, user=zabbix_user, password=zabbix_pass) zabbix = ZabbixAPI(zabbix_host, user=zabbix_user,
password=zabbix_pass, ssl_context=ssl_ctx)
else: else:
zabbix = ZabbixAPI(zabbix_host, token=zabbix_token) zabbix = ZabbixAPI(
zabbix_host, token=zabbix_token, ssl_context=ssl_ctx)
zabbix.check_auth() zabbix.check_auth()
except (APIRequestError, ProcessingError) as e: except (APIRequestError, ProcessingError) as e:
e = f"Zabbix returned the following error: {str(e)}" e = f"Zabbix returned the following error: {str(e)}"
@ -122,7 +133,8 @@ def main(arguments):
netbox_devices = list(netbox.dcim.devices.filter(**nb_device_filter)) netbox_devices = list(netbox.dcim.devices.filter(**nb_device_filter))
netbox_vms = [] netbox_vms = []
if sync_vms: if sync_vms:
netbox_vms = list(netbox.virtualization.virtual_machines.filter(**nb_vm_filter)) netbox_vms = list(
netbox.virtualization.virtual_machines.filter(**nb_vm_filter))
netbox_site_groups = convert_recordset((netbox.dcim.site_groups.all())) netbox_site_groups = convert_recordset((netbox.dcim.site_groups.all()))
netbox_regions = convert_recordset(netbox.dcim.regions.all()) netbox_regions = convert_recordset(netbox.dcim.regions.all())
netbox_journals = netbox.extras.journal_entries netbox_journals = netbox.extras.journal_entries
@ -132,7 +144,8 @@ def main(arguments):
# Set empty list for proxy processing Zabbix <= 6 # Set empty list for proxy processing Zabbix <= 6
zabbix_proxygroups = [] zabbix_proxygroups = []
if str(zabbix.version).startswith('7'): if str(zabbix.version).startswith('7'):
zabbix_proxygroups = zabbix.proxygroup.get(output=["proxy_groupid", "name"]) zabbix_proxygroups = zabbix.proxygroup.get(
output=["proxy_groupid", "name"])
# Sanitize proxy data # Sanitize proxy data
if proxy_name == "host": if proxy_name == "host":
for proxy in zabbix_proxies: for proxy in zabbix_proxies:
@ -153,7 +166,8 @@ def main(arguments):
# Check if a valid template has been found for this VM. # Check if a valid template has been found for this VM.
if not vm.zbx_template_names: if not vm.zbx_template_names:
continue continue
vm.set_hostgroup(vm_hostgroup_format,netbox_site_groups,netbox_regions) vm.set_hostgroup(vm_hostgroup_format,
netbox_site_groups, netbox_regions)
# Check if a valid hostgroup has been found for this VM. # Check if a valid hostgroup has been found for this VM.
if not vm.hostgroup: if not vm.hostgroup:
continue continue
@ -199,11 +213,13 @@ def main(arguments):
device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version, device = PhysicalDevice(nb_device, zabbix, netbox_journals, nb_version,
create_journal, logger) create_journal, logger)
logger.debug(f"Host {device.name}: started operations on device.") logger.debug(f"Host {device.name}: started operations on device.")
device.set_template(templates_config_context, templates_config_context_overrule) device.set_template(templates_config_context,
templates_config_context_overrule)
# Check if a valid template has been found for this VM. # Check if a valid template has been found for this VM.
if not device.zbx_template_names: if not device.zbx_template_names:
continue continue
device.set_hostgroup(hostgroup_format,netbox_site_groups,netbox_regions) device.set_hostgroup(
hostgroup_format, netbox_site_groups, netbox_regions)
# Check if a valid hostgroup has been found for this VM. # Check if a valid hostgroup has been found for this VM.
if not device.hostgroup: if not device.hostgroup:
continue continue