[WIP] Data uploader tool for Netbox

data_uploader/exceptions/ip_address_collision_error.py

@@ -0,0 +1,2 @@
+class IPAddressCollisionError(ValueError):
+    pass

data_uploader/exceptions/mac_address_collision_error.py

@@ -0,0 +1,2 @@
+class MacAddressCollisionError(ValueError):
+    pass

data_uploader/exceptions/missing_mandatory_param_error.py

@@ -0,0 +1,2 @@
+class MissingMandatoryParamError(ValueError):
+    pass

data_uploader/netbox_api/netbox_base.py

@@ -0,0 +1,9 @@
+from data_uploader.netbox_api.netbox_connection import NetboxApi
+
+
+class NetboxBase:
+    def __init__(self, url, token, cert):
+        self.url = url
+        self.token = token
+        self.cert = cert
+        self._netbox_api = NetboxApi.api_object(url, token, cert)

data_uploader/netbox_api/netbox_connection.py

@@ -0,0 +1,30 @@
+from data_uploader.exceptions.missing_mandatory_param_error import MissingMandatoryParamError
+import requests
+from pynetbox import api
+
+
+class NetboxApi:
+    """
+    Wraps a netbox connection as a context manager
+
+    """
+    def __init__(self):
+        pass
+
+    @staticmethod
+    def api_object(netbox_url: str, token: str, cert=None):
+        if not netbox_url:
+            raise MissingMandatoryParamError(
+                "NetBox URL is required but not provided."
+            )
+        if not token:
+            raise MissingMandatoryParamError(
+                "A token is required but not provided."
+            )
+        session = requests.Session()
+        session.verify = cert if cert else False
+
+        connection = api(url=netbox_url, token=token)
+        connection.http_session = session
+
+        return connection

data_uploader/netbox_api/netbox_dcim.py

@@ -0,0 +1,130 @@
+from typing import Union
+from data_uploader.exceptions.mac_address_collision_error import MacAddressCollisionError
+from data_uploader.netbox_api.netbox_base import NetboxBase
+from pynetbox.core.response import Record
+
+
+class NetboxDcim(NetboxBase):
+    """
+    Class for retrieving or creating DCIM objects in Netbox
+    """
+    def __init__(self, url, token, cert):
+        super().__init__(url, token, cert)
+
+    def get_device(self, hostname: str) -> Union[Record, None]:
+        """
+        Search for the name of a device (server) in Netbox by its name
+        :param hostname: name of the server to search for in Netbox
+        :return: dict of record in netbox or None if not found
+        """
+        self._netbox_api.dcim.devices.get(name=hostname)
+        return self._netbox_api.dcim.devices.get(name=hostname)
+
+    def get_device_types(self, model: str) -> Union[Record, None]:
+        """
+        Get device model type from netbox
+        :param model: model type to search for in netbox
+        :return: dict of record in netbox or None if not found
+        """
+
+        return self._netbox_api.dcim.device_types.get(slug=model)
+
+    def get_interface(self, interface_name: str, hostname: str) -> Union[Record, None]:
+        """
+        Get an interface from Netbox
+        :param interface_name: Name of interface to search for
+        :param hostname: Name of device the interface is attached to
+        """
+        return self._netbox_api.dcim.interfaces.get(name=interface_name, device=hostname)
+
+    def find_mac_addr(self, mac_address: str) -> bool:
+        """
+        Checks whether there is an interface in Netbox that already uses a specific mac address
+        :param mac_address: mac address to search Netbox for
+        :returns: Boolean indicating whether an interface with that address exists (True) or not (False)
+        """
+        check = self._netbox_api.interface.get(mac_address=mac_address)
+        return True if check else False
+
+    def create_device(self, hostname: str, site: str, location: str, tenant: str, manufacturer: str,
+                      rack: str, rack_position: int, device_type: str, serial_no: str) -> Record:
+        """
+        Create a new device in NetBox.
+
+        :param hostname: Name of device
+        :param site: Building and Room the device is located in
+        :param location: Row the device is in
+        :param tenant: Tenant for the device
+        :param manufacturer: Device manufacturer
+        :param rack: Rack the device is in
+        :param rack_position: Rack position (Optional)
+        :param device_type: Device type
+        :param serial_no: serial number (optional)
+
+        :return: Netbox Record object
+        """
+        netbox = self._netbox_api
+        netbox_device = netbox.dcim.devices.create(
+            name=hostname,
+            site=netbox.dcim.devices.get(name=site).id,
+            location=netbox.dcim.locations.get(name=location).id,
+            tenant=netbox.tenancy.tenants.get(name=tenant).id,
+            manufacturer=netbox.dcim.manufacturers.get(name=manufacturer).id,  # optional field
+            rack=netbox.dcim.racks.get(name=rack).id,
+            postion=rack_position,  # optional field - omit this field if the device is unracked
+            device_type=netbox.dcim.device_types.get(slug=device_type).id,
+            serial=serial_no,  # optional field
+            #tags=[{"name": "Tag 1"}, {"name": "Tag 2"}]  # optional field
+        )
+        return netbox_device
+
+    def create_device_type(self, model: str, manufacturer: str, slug: str, unit_height: int) -> Record:
+        """
+        Create a new device type in NetBox
+        :param model: Name of device type
+        :param manufacturer: name of manufacturer
+        :param slug: slug of model name
+        :param unit_height: height of devices of this type
+
+        :return: Netbox record of new device type
+        """
+        netbox = self._netbox_api
+        new_device_type = netbox.dcim.device_types.create(
+            manufacturer=netbox.dcim.manufacturers.get(name=manufacturer).id,
+            model=model,
+            slug=slug,
+            u_height=unit_height,
+        #    custom_fields={'cf_1': 'Custom data 1'}  # optional field
+        )
+        return new_device_type
+
+    def create_interface(self, interface_name: str, hostname: str, interface_type: str, description: str,
+                         mac_address: str) -> Record:
+        """
+        Create an interface for a device already in NetBox
+        :param interface_name: name of interface
+        :param hostname: name of device the interface is attached to
+        :param interface_type: interface type
+        :param description: description
+        :param mac_address: mac_address
+        :return: netbox_interface: Record object with details of newly created netbox interface
+        """
+        # verify whether the mac address already exists in Netbox on a specific interface
+        mac_addr_match = NetboxDcim.find_mac_addr(self, mac_address)
+
+        if mac_addr_match:
+            raise MacAddressCollisionError(
+                "MAC Address already exists in Netbox"
+            )
+
+        netbox = self._netbox_api
+
+        netbox_interface = netbox.dcim.interfaces.create(
+            name=interface_name,
+            device=netbox.dcim.devices.get(name=hostname).id,
+            type=interface_type,
+            description=description,
+            mac_address=mac_address,
+        #   tags = [{'name': 'Tag 1'}]
+        )
+        return netbox_interface

data_uploader/netbox_api/netbox_ipam.py

@@ -0,0 +1,52 @@
+from data_uploader.netbox_api.netbox_dcim import NetboxDcim
+from data_uploader.netbox_api.netbox_connection import NetboxApi
+import pynetbox
+
+class NetboxIpam:
+    def __init__(self, url, token, cert):
+        self.url = url
+        self.token = token
+        self. cert = cert
+        self._netbox_api = NetboxApi.api_object(url, token, cert)
+
+    def search_mac_address(self, mac_addr: str):
+        """
+        Search Netbox for a mac address assigned to any interface
+        This will call a method from NetBox Dcim to do this
+        :return: Record object with interface the mac address is associated with or None if not found
+        """
+        netbox = self._netbox_api
+        interface = netbox.dcim.interfaces.get(mac_address=mac_addr)
+
+        return interface
+
+    def get_ip_address(self, ip_addr: str):
+        """
+        Search netbox for an IP Address
+        :param ip_addr: IP address to search for in Netbox
+        :return: Record object with IP Address or None if not found
+        """
+        netbox = self._netbox_api
+        return netbox.ipam.ip_addresses.get(address=ip_addr)
+
+    def create_ip_address(self, hostname: str, interface:str, ip_address: str, tenant: str):
+        """
+        Create a new IP address in Netbox
+        :param hostname: Name of device in Netbox (must be present in Netbox)
+        :param interface: Interface attached to device (must be present in Netbox)
+        :param ip_address: IP address to create
+        :param tenant:  Tenant IP address belongs to (must be present in Netbox)
+        """
+
+        interface = NetboxDcim.get_interface(name=interface, device=hostname)
+        netbox_ip = netbox.ipam.ip_addresses.create(
+            address="ip-address",
+            tenant=netbox.tenancy.tenants.get(name='tenant-name').id,
+            tags=[{'name': 'Tag 1'}],
+        )
+        # assign IP Address to device's network interface
+        netbox_ip.assigned_object = interface
+        netbox_ip.assigned_object_id = interface.id
+        netbox_ip.assigned_object_type = 'dcim.interface'
+        # save changes to IP record in Netbox
+        netbox_ip.save()

data_uploader/requirements.txt

@@ -0,0 +1,6 @@
+pynetbox
+pytest
+argparse
+logger
+requests
+nose

data_uploader/system_data_class.py

@@ -0,0 +1,14 @@
+from dataclasses import dataclass
+
+@dataclass
+class SystemDataClass:
+    """
+    Class for keeping track of server properties
+    """
+    manufacturer: str
+    model: str
+    primaryMAC: str
+    hostname: str
+    serviceTag: str
+    ipmiIPAddress: str
+

data_uploader/test/test_netbox_connection.py

@@ -0,0 +1,55 @@
+import unittest
+from unittest.mock import NonCallableMock, patch
+from nose.tools import raises
+from data_uploader.exceptions.missing_mandatory_param_error import MissingMandatoryParamError
+from data_uploader.netbox_api.netbox_connection import NetboxApi
+
+
+class NetboxApiTests(unittest.TestCase):
+    @staticmethod
+    @patch('data_uploader.netbox_api.netbox_connection.api')
+    def test_api_object_create(mock_api):
+        """
+        Tests that we do get an API object an
+        """
+        mock_url = "example.com"
+        mock_token = NonCallableMock()
+        mock_cert = NonCallableMock()
+
+        api = NetboxApi.api_object(mock_url, mock_token, mock_cert)
+        assert mock_api.return_value == api
+
+    @staticmethod
+    @patch('data_uploader.netbox_api.netbox_connection.api')
+    def test_api_no_cert(mock_api):
+        """
+        Test that an api object is created even when we don't have a path for a certificate
+        """
+        mock_url = "example.com"
+        mock_token = NonCallableMock()
+        mock_cert = None
+
+        api = NetboxApi.api_object(mock_url, mock_token, mock_cert)
+        assert mock_api.return_value == api
+
+    @raises(MissingMandatoryParamError)
+    def test_api_throws_for_no_url(self):
+        """
+        Tests a None type will throw error if used as url
+        """
+        missing_url = None
+        mock_token = NonCallableMock()
+        mock_cert = NonCallableMock()
+        api = NetboxApi.api_object(missing_url, mock_token, mock_cert)
+        api.assertRaises(MissingMandatoryParamError)
+
+    @raises(MissingMandatoryParamError)
+    def test_api_throws_for_no_token(self):
+        """
+        Tests a None type will throw if used as token
+        """
+        mock_url = NonCallableMock()
+        missing_token = None
+        mock_cert = NonCallableMock()
+        NetboxApi.api_object(mock_url, missing_token, mock_cert)
+        self.assertRaises(MissingMandatoryParamError)