adds support for 'primary_mac_address' in NetBox 4.2

module/netbox/__init__.py

@@ -34,6 +34,7 @@
     NBVirtualDisk,
     NBInterface,
     NBIPAddress,
+    NBMACAddress,
     NBFHRPGroupItem,
     NBInventoryItem,
     NBPowerPort

module/netbox/connection.py

@@ -650,16 +650,17 @@ def update_object(self, nb_object_sub_class, unset=False, last_run=False):
                         # resolve dependency issues in last run
                         # primary IP always set in last run
                         if value.get_nb_reference() is None or \
-                                (key.startswith("primary_ip") and last_run is False):
+                                (key.startswith("primary_ip") and last_run is False) or \
+                                (key.startswith("primary_mac_address") and last_run is False):
                             unresolved_dependency_data[key] = value
                         else:
                             data_to_patch[key] = value.get_nb_reference()
 
                     else:
                         data_to_patch[key] = value
 
-            # special case for IP address
-            if isinstance(this_object, NBIPAddress):
+            # special case for IP and MAC addresses
+            if isinstance(this_object, (NBIPAddress, NBMACAddress)):
                 # if object is new and has no id, then we need to remove assigned_object_type from data_to_patch
                 if "assigned_object_id" in unresolved_dependency_data.keys() and \
                         "assigned_object_type" in data_to_patch.keys():

module/netbox/object_classes.py

@@ -1687,6 +1687,7 @@ def __init__(self, *args, **kwargs):
             "virtual_machine": NBVM,
             "enabled": bool,
             "mac_address": str,
+            "primary_mac_address": NBMACAddress,
             "mtu": int,
             "mode": ["access", "tagged", "tagged-all"],
             "untagged_vlan": NBVLAN,
@@ -1722,6 +1723,7 @@ def __init__(self, *args, **kwargs):
             "type": NetBoxInterfaceType().get_netbox_type_list(),
             "enabled": bool,
             "mac_address": str,
+            "primary_mac_address": NBMACAddress,
             "wwn": str,
             "mgmt_only": bool,
             "mtu": int,
@@ -1913,6 +1915,116 @@ def remove_interface_association(self):
         if o_type is not None:
             self.unset_attribute("assigned_object_type")
 
+class NBMACAddress(NetBoxObject):
+    name = "MAC address"
+    api_path = "dcim/mac-addresses"
+    primary_key = "mac_address"
+    prune = True
+    min_netbox_version = "4.2"
+
+    def __init__(self, *args, **kwargs):
+        self.data_model = {
+            "mac_address": str,
+            "assigned_object_type": ["dcim.interface", "virtualization.vminterface"],
+            "assigned_object_id": [NBInterface, NBVMInterface],
+            "description": 200,
+            "tags": NBTagList,
+        }
+
+        # add relation between two attributes
+        self.data_model_relation = {
+            "dcim.interface": NBInterface,
+            "virtualization.vminterface": NBVMInterface,
+            NBInterface: "dcim.interface",
+            NBVMInterface: "virtualization.vminterface",
+        }
+        super().__init__(*args, **kwargs)
+
+    def resolve_relations(self):
+
+        o_id = self.data.get("assigned_object_id")
+        o_type = self.data.get("assigned_object_type")
+
+        # this needs special treatment as the object type depends on a second model key
+        if o_type is not None and o_type not in self.data_model.get("assigned_object_type"):
+
+            log.error(f"Attribute 'assigned_object_type' for '{self.get_display_name()}' invalid: {o_type}")
+            do_error_exit(f"Error while resolving relations for {self.get_display_name()}")
+
+        if isinstance(o_id, int) and o_type is not None:
+            self.data["assigned_object_id"] = self.inventory.get_by_id(self.data_model_relation.get(o_type), nb_id=o_id)
+
+        super().resolve_relations()
+
+
+    def update(self, data=None, read_from_netbox=False, source=None):
+
+        object_type = data.get("assigned_object_type")
+        assigned_object = data.get("assigned_object_id")
+
+        # we got an object data structure where we have to find the object
+        if read_from_netbox is False and assigned_object is not None:
+
+            if not isinstance(assigned_object, NetBoxObject):
+
+                data["assigned_object_id"] = \
+                    self.inventory.add_update_object(self.data_model_relation.get(object_type), data=assigned_object)
+
+            else:
+                # noinspection PyTypeChecker
+                data["assigned_object_type"] = self.data_model_relation.get(type(assigned_object))
+
+        super().update(data=data, read_from_netbox=read_from_netbox, source=source)
+
+        # we need to tell NetBox which object type this is meant to be
+        if "assigned_object_id" in self.updated_items:
+            self.updated_items.append("assigned_object_type")
+
+        # if ip association has been removed we also need to get rid of object type
+        if "assigned_object_type" in self.updated_items and self.data.get("assigned_object_id") is None \
+                and "assigned_object_type" in self.updated_items:
+            self.updated_items.remove("assigned_object_type")
+
+    def get_interface(self):
+        o_id = self.data.get("assigned_object_id")
+        o_type = self.data.get("assigned_object_type")
+
+        if isinstance(o_id, (NBInterface, NBVMInterface)):
+            return o_id
+
+        if o_type is None or not isinstance(o_id, int):
+            return
+
+        if o_type not in self.data_model.get("assigned_object_type"):
+            return
+
+        return self.inventory.get_by_id(self.data_model_relation.get(o_type), nb_id=o_id)
+
+    def get_device_vm(self):
+
+        o_interface = self.get_interface()
+
+        if o_interface is None:
+            return
+
+        if isinstance(o_interface, NBInterface):
+            return o_interface.data.get("device")
+        elif isinstance(o_interface, NBVMInterface):
+            return o_interface.data.get("virtual_machine")
+
+    def remove_interface_association(self):
+        o_id = self.data.get("assigned_object_id")
+        o_type = self.data.get("assigned_object_type")
+        o_device = self.get_device_vm()
+
+        if grab(o_device, "data.primary_mac_address") is self:
+            o_device.unset_attribute("primary_mac_address")
+
+        if o_id is not None:
+            self.unset_attribute("assigned_object_id")
+        if o_type is not None:
+            self.unset_attribute("assigned_object_type")
+
 
 class NBFHRPGroupItem(NetBoxObject):
     """

module/sources/common/source_base.py

@@ -11,6 +11,7 @@
 
 from ipaddress import ip_interface, ip_address, IPv6Address, IPv4Address, IPv6Network, IPv4Network
 from typing import List
+from packaging import version
 
 from module.netbox import *
 from module.common.logging import get_logger
@@ -266,6 +267,12 @@ def add_update_interface(self, interface_object, device_object, interface_data,
             added to this interface
         """
 
+        # handle change to mac_address object from NetBox 4.2 on
+        interface_mac_address = None
+        if version.parse(self.inventory.netbox_api_version) >= version.parse("4.2.0"):
+            interface_mac_address = interface_data.get("mac_address")
+            del(interface_data["mac_address"])
+
         ip_tenant_inheritance_order = self.settings.ip_tenant_inheritance_order
 
         if not isinstance(interface_data, dict):
@@ -312,6 +319,32 @@ def add_update_interface(self, interface_object, device_object, interface_data,
         else:
             interface_object.update(data=interface_data, source=self)
 
+        if version.parse(self.inventory.netbox_api_version) >= version.parse("4.2.0") and \
+                interface_mac_address is not None :
+
+            primary_mac_address_object = grab(interface_object, "data.primary_mac_address")
+
+            if primary_mac_address_object is None or grab(primary_mac_address_object, "data.mac_address") != interface_mac_address:
+
+                primary_mac_address_object = None
+                for mac_address_object in self.inventory.get_all_items(NBMACAddress):
+                    if grab(mac_address_object, "data.mac_address") == interface_mac_address and grab(mac_address_object, "data.assigned_object_id") is None:
+                        primary_mac_address_object = mac_address_object
+                        break
+
+                primary_mac_address_data = {
+                        "mac_address": interface_mac_address,
+                        "assigned_object_id": interface_object,
+                        "assigned_object_type": interface_class
+                    }
+
+                if primary_mac_address_object is None:
+                    primary_mac_address_object = self.inventory.add_object(NBMACAddress, data=primary_mac_address_data, source=self)
+                else:
+                    primary_mac_address_object.update(data=primary_mac_address_data, source=self)
+
+            interface_object.update(data={"primary_mac_address": primary_mac_address_object}, source=self)
+
         ip_address_objects = list()
         matching_ip_prefixes = list()
         # add all interface IPs
@@ -686,8 +719,6 @@ def add_vlan_group(self, vlan_data, vlan_site) -> NBVLAN | dict:
         else:
             log.debug2("No matching VLAN group found")
 
-        print(vlan_data)
-
         return vlan_data
 
     def get_vlan_object_if_exists(self, vlan_data=None, vlan_site=None):

module/sources/vmware/connection.py

@@ -73,7 +73,8 @@ class VMWareHandler(SourceBase):
         NBVLAN,
         NBVLANGroup,
         NBCustomField,
-        NBVirtualDisk
+        NBVirtualDisk,
+        NBMACAddress
     ]
 
     source_type = "vmware"