diff --git a/development/creds.example.env b/development/creds.example.env index 74b9371af..780d04b29 100644 --- a/development/creds.example.env +++ b/development/creds.example.env @@ -44,3 +44,5 @@ NAUTOBOT_APIC_VERIFY_DEVNET=False # NAUTOBOT_APIC_SITE_DEVNET="DevNet Sandbox" SERVICENOW_PASSWORD="changeme" + +IPFABRIC_API_TOKEN=secrettoken diff --git a/development/development.env b/development/development.env index 72e993940..19a0d075e 100644 --- a/development/development.env +++ b/development/development.env @@ -39,7 +39,10 @@ MYSQL_USER=${NAUTOBOT_DB_USER} MYSQL_DATABASE=${NAUTOBOT_DB_NAME} MYSQL_ROOT_HOST=% -NAUTOBOT_CELERY_TASK_SOFT_TIME_LIMIT=600 +NAUTOBOT_HOST="http://nautobot:8080" + +NAUTOBOT_CELERY_TASK_SOFT_TIME_LIMIT=3600 +NAUTOBOT_CELERY_TASK_TIME_LIMIT=3600 NAUTOBOT_SSOT_HIDE_EXAMPLE_JOBS="False" @@ -83,3 +86,8 @@ NAUTOBOT_SSOT_INFOBLOX_VERIFY_SSL="True" NAUTOBOT_SSOT_ENABLE_SERVICENOW="True" SERVICENOW_INSTANCE="" SERVICENOW_USERNAME="" + +NAUTOBOT_SSOT_ENABLE_IPFABRIC="True" +IPFABRIC_HOST="https://ipfabric.example.com" +IPFABRIC_SSL_VERIFY="True" +IPFABRIC_TIMEOUT=15 diff --git a/development/nautobot_config.py b/development/nautobot_config.py index 471988aca..264c29352 100644 --- a/development/nautobot_config.py +++ b/development/nautobot_config.py @@ -135,11 +135,22 @@ PLUGINS = [ "nautobot_ssot", # "nautobot_device_lifecycle_mgmt", + # Enable chatops after Nautobot v2 compatible release + # "nautobot_chatops", ] # Plugins configuration settings. These settings are used by various plugins that the user may have installed. # Each key in the dictionary is the name of an installed plugin and its value is a dictionary of settings. PLUGINS_CONFIG = { + # Enable chatops after Nautobot v2 compatible release + # "nautobot_chatops": { + # "enable_slack": True, + # "slack_api_token": os.getenv("SLACK_API_TOKEN"), + # "slack_signing_secret": os.getenv("SLACK_SIGNING_SECRET"), + # "session_cache_timeout": 3600, + # "ipfabric_api_token": os.getenv("IPFABRIC_API_TOKEN"), + # "ipfabric_host": os.getenv("IPFABRIC_HOST"), + # }, "nautobot_ssot": { # URL and credentials should be configured as environment variables on the host system "aci_apics": {x: os.environ[x] for x in os.environ if "APIC" in x}, @@ -190,6 +201,7 @@ "enable_aci": is_truthy(os.getenv("NAUTOBOT_SSOT_ENABLE_ACI")), "enable_aristacv": is_truthy(os.getenv("NAUTOBOT_SSOT_ENABLE_ARISTACV")), "enable_infoblox": is_truthy(os.getenv("NAUTOBOT_SSOT_ENABLE_INFOBLOX")), + "enable_ipfabric": is_truthy(os.getenv("NAUTOBOT_SSOT_ENABLE_IPFABRIC")), "enable_servicenow": is_truthy(os.getenv("NAUTOBOT_SSOT_ENABLE_SERVICENOW")), "hide_example_jobs": is_truthy(os.getenv("NAUTOBOT_SSOT_HIDE_EXAMPLE_JOBS")), "infoblox_default_status": os.getenv("NAUTOBOT_SSOT_INFOBLOX_DEFAULT_STATUS", "active"), @@ -206,6 +218,11 @@ "infoblox_username": os.getenv("NAUTOBOT_SSOT_INFOBLOX_USERNAME"), "infoblox_verify_ssl": is_truthy(os.getenv("NAUTOBOT_SSOT_INFOBLOX_VERIFY_SSL", True)), "infoblox_wapi_version": os.getenv("NAUTOBOT_SSOT_INFOBLOX_WAPI_VERSION", "v2.12"), + "ipfabric_api_token": os.getenv("IPFABRIC_API_TOKEN"), + "ipfabric_host": os.getenv("IPFABRIC_HOST"), + "ipfabric_ssl_verify": is_truthy(os.getenv("IPFABRIC_VERIFY", "False")), + "ipfabric_timeout": int(os.getenv("IPFABRIC_TIMEOUT", "15")), + "nautobot_host": os.getenv("NAUTOBOT_HOST"), "servicenow_instance": os.getenv("SERVICENOW_INSTANCE", ""), "servicenow_password": os.getenv("SERVICENOW_PASSWORD", ""), "servicenow_username": os.getenv("SERVICENOW_USERNAME", ""), diff --git a/nautobot_ssot/__init__.py b/nautobot_ssot/__init__.py index 4c16ef036..f827a082e 100644 --- a/nautobot_ssot/__init__.py +++ b/nautobot_ssot/__init__.py @@ -55,6 +55,7 @@ class NautobotSSOTPluginConfig(PluginConfig): "enable_aci": False, "enable_aristacv": False, "enable_infoblox": False, + "enable_ipfabric": False, "enable_servicenow": False, "hide_example_jobs": True, "infoblox_default_status": "", @@ -70,6 +71,11 @@ class NautobotSSOTPluginConfig(PluginConfig): "infoblox_username": "", "infoblox_verify_ssl": True, "infoblox_wapi_version": "", + "ipfabric_api_token": "", + "ipfabric_host": "", + "ipfabric_ssl_verify": True, + "ipfabric_timeout": 15, + "ipfabric_nautobot_host": "", "servicenow_instance": "", "servicenow_password": "", "servicenow_username": "", diff --git a/nautobot_ssot/integrations/ipfabric/__init__.py b/nautobot_ssot/integrations/ipfabric/__init__.py new file mode 100644 index 000000000..1d5cf9e79 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/__init__.py @@ -0,0 +1 @@ +"""Base module for IPFabric integration.""" diff --git a/nautobot_ssot/integrations/ipfabric/constants.py b/nautobot_ssot/integrations/ipfabric/constants.py new file mode 100644 index 000000000..bf8e32e6d --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/constants.py @@ -0,0 +1,15 @@ +"""Constants used by IPFabric Integration.""" + +ALLOW_DUPLICATE_ADDRESSES = True +DEFAULT_DEVICE_ROLE = "Network Device" +DEFAULT_DEVICE_ROLE_COLOR = "ff0000" +DEFAULT_DEVICE_STATUS = "Active" +DEFAULT_DEVICE_STATUS_COLOR = "ff0000" +DEFAULT_INTERFACE_MAC = "00:00:00:00:00:01" +DEFAULT_INTERFACE_MTU = 1500 +DEFAULT_INTERFACE_TYPE = "1000base-t" +SAFE_DELETE_DEVICE_STATUS = "Deprecated" +SAFE_DELETE_IPADDRESS_STATUS = "Deprecated" +SAFE_DELETE_LOCATION_STATUS = "Decommissioning" +SAFE_DELETE_VLAN_STATUS = "Inventory" +SAFE_IPADDRESS_INTERFACES_STATUS = "Deprecated" diff --git a/nautobot_ssot/integrations/ipfabric/diffsync/__init__.py b/nautobot_ssot/integrations/ipfabric/diffsync/__init__.py new file mode 100644 index 000000000..4c6f15145 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/diffsync/__init__.py @@ -0,0 +1,4 @@ +"""Diffsync Utilities, Adapters and Models.""" +from .adapters_shared import DiffSyncModelAdapters + +__all__ = ("DiffSyncModelAdapters",) diff --git a/nautobot_ssot/integrations/ipfabric/diffsync/adapter_ipfabric.py b/nautobot_ssot/integrations/ipfabric/diffsync/adapter_ipfabric.py new file mode 100644 index 000000000..d3e761e9b --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/diffsync/adapter_ipfabric.py @@ -0,0 +1,158 @@ +# pylint: disable=duplicate-code +"""DiffSync adapter class for Ip Fabric.""" + +import logging + +from diffsync import ObjectAlreadyExists +from nautobot.dcim.models import Device +from nautobot.ipam.models import VLAN +from netutils.mac import mac_to_format + +from nautobot_ssot.integrations.ipfabric.constants import ( + DEFAULT_INTERFACE_TYPE, + DEFAULT_INTERFACE_MTU, + DEFAULT_INTERFACE_MAC, + DEFAULT_DEVICE_ROLE, + DEFAULT_DEVICE_STATUS, +) +from nautobot_ssot.integrations.ipfabric.diffsync import DiffSyncModelAdapters + +logger = logging.getLogger("nautobot.jobs") + +device_serial_max_length = Device._meta.get_field("serial").max_length +name_max_length = VLAN._meta.get_field("name").max_length + + +class IPFabricDiffSync(DiffSyncModelAdapters): + """Nautobot adapter for DiffSync.""" + + def __init__(self, job, sync, client, *args, **kwargs): + """Initialize the NautobotDiffSync.""" + super().__init__(*args, **kwargs) + self.job = job + self.sync = sync + self.client = client + + def load_sites(self): + """Add IP Fabric Location objects as DiffSync Location models.""" + sites = self.client.inventory.sites.all() + for site in sites: + try: + location = self.location(diffsync=self, name=site["siteName"], site_id=site["id"], status="Active") + self.add(location) + except ObjectAlreadyExists: + logger.warning(f"Duplicate Location discovered, {site}") + + def load_device_interfaces(self, device_model, interfaces, device_primary_ip): + """Create and load DiffSync Interface model objects for a specific device.""" + device_interfaces = [iface for iface in interfaces if iface.get("hostname") == device_model.name] + pseudo_interface = pseudo_management_interface(device_model.name, device_interfaces, device_primary_ip) + + if pseudo_interface: + device_interfaces.append(pseudo_interface) + logger.info("Pseudo MGMT Interface: %s", pseudo_interface) + + for iface in device_interfaces: + ip_address = iface.get("primaryIp") + try: + interface = self.interface( + diffsync=self, + name=iface.get("intName"), + device_name=iface.get("hostname"), + description=iface.get("dscr", ""), + enabled=True, + mac_address=mac_to_format(iface.get("mac"), "MAC_COLON_TWO").upper() + if iface.get("mac") + else DEFAULT_INTERFACE_MAC, + mtu=iface.get("mtu") if iface.get("mtu") else DEFAULT_INTERFACE_MTU, + type=DEFAULT_INTERFACE_TYPE, + mgmt_only=iface.get("mgmt_only", False), + ip_address=ip_address, + # TODO: why is only IPv4? and why /32? + subnet_mask="255.255.255.255", + ip_is_primary=ip_address is not None and ip_address == device_primary_ip, + status="Active", + ) + self.add(interface) + device_model.add_child(interface) + except ObjectAlreadyExists: + logger.warning(f"Duplicate Interface discovered, {iface}") + + def load(self): + """Load data from IP Fabric.""" + self.load_sites() + devices = self.client.inventory.devices.all() + interfaces = self.client.inventory.interfaces.all() + vlans = self.client.fetch_all("tables/vlan/site-summary") + + for location in self.get_all(self.location): + if location.name is None: + continue + location_vlans = [vlan for vlan in vlans if vlan["siteName"] == location.name] + for vlan in location_vlans: + if not vlan["vlanId"] or (vlan["vlanId"] < 1 or vlan["vlanId"] > 4094): + logger.warning( + f"Not syncing VLAN, NAME: {vlan.get('vlanName')} due to invalid VLAN ID: {vlan.get('vlanId')}." + ) + continue + description = vlan.get("dscr") if vlan.get("dscr") else f"VLAN ID: {vlan['vlanId']}" + vlan_name = vlan.get("vlanName") if vlan.get("vlanName") else f"{vlan['siteName']}:{vlan['vlanId']}" + if len(vlan_name) > name_max_length: + logger.warning(f"Not syncing VLAN, {vlan_name} due to character limit exceeding {name_max_length}.") + continue + try: + vlan = self.vlan( + diffsync=self, + name=vlan_name, + location=vlan["siteName"], + vid=vlan["vlanId"], + status="Active", + description=description, + ) + self.add(vlan) + location.add_child(vlan) + except ObjectAlreadyExists: + logger.warning(f"Duplicate VLAN discovered, {vlan}") + + location_devices = [device for device in devices if device["siteName"] == location.name] + for device in location_devices: + device_primary_ip = device["loginIp"] + sn_length = len(device["sn"]) + serial_number = device["sn"] if sn_length < device_serial_max_length else "" + if not serial_number: + logger.warning( + ( + f"Serial Number will not be recorded for {device['hostname']} due to character limit. " + f"{sn_length} exceeds {device_serial_max_length}" + ) + ) + try: + device_model = self.device( + diffsync=self, + name=device["hostname"], + location_name=device["siteName"], + model=device.get("model") if device.get("model") else f"Default-{device.get('vendor')}", + vendor=device.get("vendor").capitalize(), + serial_number=serial_number, + role=device.get("devType") if device.get("devType") else DEFAULT_DEVICE_ROLE, + status=DEFAULT_DEVICE_STATUS, + ) + self.add(device_model) + location.add_child(device_model) + self.load_device_interfaces(device_model, interfaces, device_primary_ip) + except ObjectAlreadyExists: + logger.warning(f"Duplicate Device discovered, {device}") + + +def pseudo_management_interface(hostname, device_interfaces, device_primary_ip): + """Return a dict for an non-existing interface for NAT management addresses.""" + if any(iface for iface in device_interfaces if iface.get("primaryIp", "") == device_primary_ip): + return None + return { + "hostname": hostname, + "intName": "pseudo_mgmt", + "dscr": "pseudo interface for NAT IP address", + "primaryIp": device_primary_ip, + "type": "virtual", + "mgmt_only": True, + } diff --git a/nautobot_ssot/integrations/ipfabric/diffsync/adapter_nautobot.py b/nautobot_ssot/integrations/ipfabric/diffsync/adapter_nautobot.py new file mode 100644 index 000000000..e2d45a681 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/diffsync/adapter_nautobot.py @@ -0,0 +1,242 @@ +# pylint: disable=duplicate-code +# pylint: disable=too-many-arguments +# Load method is packed with conditionals # pylint: disable=too-many-branches +"""DiffSync adapter class for Nautobot as source-of-truth.""" +from collections import defaultdict +from typing import Any, ClassVar, List, Optional +import logging + +from diffsync import DiffSync +from diffsync.exceptions import ObjectAlreadyExists +from django.db import IntegrityError, transaction +from django.db.models import ProtectedError, Q +from nautobot.dcim.models import Device, Location +from nautobot.extras.models import Tag +from nautobot.ipam.models import VLAN, Interface +from nautobot.core.choices import ColorChoices +from netutils.mac import mac_to_format + +from nautobot_ssot.integrations.ipfabric.diffsync import DiffSyncModelAdapters + +from nautobot_ssot.integrations.ipfabric.constants import ( + DEFAULT_INTERFACE_TYPE, + DEFAULT_INTERFACE_MTU, + DEFAULT_INTERFACE_MAC, +) + +logger = logging.getLogger("nautobot.ssot.ipfabric") + + +class NautobotDiffSync(DiffSyncModelAdapters): + """Nautobot adapter for DiffSync.""" + + objects_to_delete = defaultdict(list) + + _vlan: ClassVar[Any] = VLAN + _device: ClassVar[Any] = Device + _location: ClassVar[Any] = Location + _interface: ClassVar[Any] = Interface + + def __init__( + self, + job, + sync, + sync_ipfabric_tagged_only: bool, + location_filter: Optional[Location], + *args, + **kwargs, + ): + """Initialize the NautobotDiffSync.""" + super().__init__(*args, **kwargs) + self.job = job + self.sync = sync + self.sync_ipfabric_tagged_only = sync_ipfabric_tagged_only + self.location_filter = location_filter + + def sync_complete(self, source: DiffSync, *args, **kwargs): + """Clean up function for DiffSync sync. + + Once the sync is complete, this function runs deleting any objects + from Nautobot that need to be deleted in a specific order. + + Args: + source (DiffSync): DiffSync + """ + for grouping in ( + "_vlan", + "_interface", + "_device", + "_location", + ): + for nautobot_object in self.objects_to_delete[grouping]: + if NautobotDiffSync.safe_delete_mode: + continue + try: + nautobot_object.delete() + except ProtectedError: + logger.warning("Deletion failed protected object", extra={"object": nautobot_object}) + except IntegrityError: + logger.warning(f"Deletion failed due to IntegrityError with {nautobot_object}") + + self.objects_to_delete[grouping] = [] + return super().sync_complete(source, *args, **kwargs) + + def load_interfaces(self, device_record: Device, diffsync_device): + """Import a single Nautobot Interface object as a DiffSync Interface model.""" + device_primary_ip = None + if device_record.primary_ip4: + device_primary_ip = device_record.primary_ip4 + elif device_record.primary_ip6: + device_primary_ip = device_record.primary_ip6 + + for interface_record in device_record.interfaces.all(): + interface = self.interface( + diffsync=self, + status=device_record.status.name, + name=interface_record.name, + device_name=device_record.name, + description=interface_record.description if interface_record.description else None, + enabled=True, + mac_address=mac_to_format(str(interface_record.mac_address), "MAC_COLON_TWO").upper() + if interface_record.mac_address + else DEFAULT_INTERFACE_MAC, + subnet_mask="255.255.255.255", + mtu=interface_record.mtu if interface_record.mtu else DEFAULT_INTERFACE_MTU, + type=DEFAULT_INTERFACE_TYPE, + mgmt_only=interface_record.mgmt_only if interface_record.mgmt_only else False, + pk=interface_record.pk, + ip_is_primary=interface_record.ip_addresses.first() == device_primary_ip + if device_primary_ip + else False, + ip_address=str(interface_record.ip_addresses.first().host) + if interface_record.ip_addresses.first() + else None, + ) + self.add(interface) + diffsync_device.add_child(interface) + + def load_device(self, filtered_devices: List, location): + """Load Devices from Nautobot.""" + for device_record in filtered_devices: + if self.job.debug: + logger.debug("Loading Nautobot Device: %s", device_record.name) + device = self.device( + diffsync=self, + name=device_record.name, + model=str(device_record.device_type), + role=str(device_record.role.cf.get("ipfabric_type")) + if str(device_record.role.cf.get("ipfabric_type")) + else str(device_record.role), + location_name=device_record.location.name, + vendor=str(device_record.device_type.manufacturer), + status=device_record.status.name, + serial_number=device_record.serial if device_record.serial else "", + ) + try: + self.add(device) + except ObjectAlreadyExists: + logger.warning(f"Duplicate device discovered, {device_record.name}") + continue + + location.add_child(device) + self.load_interfaces(device_record=device_record, diffsync_device=device) + + def load_vlans(self, filtered_vlans: List, location): + """Add Nautobot VLAN objects as DiffSync VLAN models.""" + for vlan_record in filtered_vlans: + if not vlan_record: + continue + vlan = self.vlan( + diffsync=self, + name=vlan_record.name, + location=vlan_record.location.name, + status=vlan_record.status.name if vlan_record.status else "Active", + vid=vlan_record.vid, + vlan_pk=vlan_record.pk, + description=vlan_record.description, + ) + try: + self.add(vlan) + except ObjectAlreadyExists: + logger.warning(f"Duplicate VLAN discovered, {vlan_record.name}") + continue + location.add_child(vlan) + + def get_initial_location(self, ssot_tag: Tag): + """Identify the location objects based on user defined job inputs. + + Args: + ssot_tag (Tag): Tag used for filtering + """ + # Simple check / validate Tag is present. + if self.sync_ipfabric_tagged_only: + location_objects = Location.objects.filter(tags__name=ssot_tag.name) + if self.location_filter: + location_objects = Location.objects.filter( + Q(name=self.location_filter.name) & Q(tags__name=ssot_tag.name) + ) + if not location_objects: + logger.warning( + f"{self.location_filter.name} was used to filter, alongside SSoT Tag. {self.location_filter.name} is not tagged." + ) + elif not self.sync_ipfabric_tagged_only: + if self.location_filter: + location_objects = Location.objects.filter(name=self.location_filter.name) + else: + location_objects = Location.objects.all() + return location_objects + + @transaction.atomic + def load_data(self): + """Add Nautobot Location objects as DiffSync Location models.""" + ssot_tag, _ = Tag.objects.get_or_create( + name="SSoT Synced from IPFabric", + defaults={ + "description": "Object synced at some point from IPFabric to Nautobot", + "color": ColorChoices.COLOR_LIGHT_GREEN, + }, + ) + location_objects = self.get_initial_location(ssot_tag) + # The parent object that stores all children, is the Location. + if self.job.debug: + logger.debug("Found %s Nautobot Location objects to start sync from", location_objects.count()) + + if location_objects: + for location_record in location_objects: + try: + location = self.location( + diffsync=self, + name=location_record.name, + site_id=location_record.custom_field_data.get("ipfabric-site-id"), + status=location_record.status.name, + ) + except AttributeError: + logger.error( + "Error loading %s, invalid or missing attributes on object. Skipping...", location_record + ) + continue + self.add(location) + try: + # Load Location's Children - Devices with Interfaces, if any. + if self.sync_ipfabric_tagged_only: + nautobot_location_devices = Device.objects.filter( + Q(location=location_record) & Q(tags__name=ssot_tag.name) + ) + else: + nautobot_location_devices = Device.objects.filter(location=location_record) + if nautobot_location_devices.exists(): + self.load_device(nautobot_location_devices, location) + + # Load Location Children - Vlans, if any. + nautobot_location_vlans = VLAN.objects.filter(location=location_record) + if not nautobot_location_vlans.exists(): + continue + self.load_vlans(nautobot_location_vlans, location) + except Location.DoesNotExist: + logger.error("Unable to find Location, %s.", location_record) + else: + logger.warning("No Nautobot records to load.") + + def load(self): + """Load data from Nautobot.""" + self.load_data() diff --git a/nautobot_ssot/integrations/ipfabric/diffsync/adapters_shared.py b/nautobot_ssot/integrations/ipfabric/diffsync/adapters_shared.py new file mode 100644 index 000000000..5f0d86a44 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/diffsync/adapters_shared.py @@ -0,0 +1,22 @@ +"""Diff sync shared adapter class attritbutes to synchronize applications.""" + +from typing import ClassVar + +from diffsync import DiffSync + +from nautobot_ssot.integrations.ipfabric.diffsync import diffsync_models + + +class DiffSyncModelAdapters(DiffSync): + """Nautobot adapter for DiffSync.""" + + safe_delete_mode: ClassVar[bool] = True + + location = diffsync_models.Location + device = diffsync_models.Device + interface = diffsync_models.Interface + vlan = diffsync_models.Vlan + + top_level = [ + "location", + ] diff --git a/nautobot_ssot/integrations/ipfabric/diffsync/diffsync_models.py b/nautobot_ssot/integrations/ipfabric/diffsync/diffsync_models.py new file mode 100644 index 000000000..104cc1fb4 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/diffsync/diffsync_models.py @@ -0,0 +1,456 @@ +# pylint: disable=duplicate-code +# Ignore return statements for updates and deletes, # pylint:disable=R1710 +# Ignore too many args # pylint:disable=too-many-locals +"""DiffSyncModel subclasses for Nautobot-to-IPFabric data sync.""" +from typing import Any, ClassVar, List, Optional +from uuid import UUID +import logging + +from diffsync import DiffSyncModel +from django.core.exceptions import ValidationError +from django.db.models import Q +from nautobot.dcim.models import Device as NautobotDevice +from nautobot.dcim.models import DeviceType, Location as NautobotLocation +from nautobot.extras.models import Role, Tag +from nautobot.extras.models.statuses import Status +from nautobot.ipam.models import VLAN +from nautobot.core.choices import ColorChoices + +import nautobot_ssot.integrations.ipfabric.utilities.nbutils as tonb_nbutils +from nautobot_ssot.integrations.ipfabric.constants import ( + DEFAULT_DEVICE_ROLE, + DEFAULT_DEVICE_ROLE_COLOR, + DEFAULT_DEVICE_STATUS, + DEFAULT_DEVICE_STATUS_COLOR, + DEFAULT_INTERFACE_MAC, + SAFE_DELETE_LOCATION_STATUS, + SAFE_DELETE_DEVICE_STATUS, + SAFE_DELETE_IPADDRESS_STATUS, + SAFE_DELETE_VLAN_STATUS, +) + +logger = logging.getLogger(__name__) + + +class DiffSyncExtras(DiffSyncModel): + """Additional components to mix and subclass from with `DiffSyncModel`.""" + + safe_delete_mode: ClassVar[bool] = True + + def safe_delete(self, nautobot_object: Any, safe_delete_status: Optional[str] = None): + """Safe delete an object, by adding tags or changing it's default status. + + Args: + nautobot_object (Any): Any type of Nautobot object + safe_delete_status (Optional[str], optional): Status name, optional as some objects don't have status field. Defaults to None. + """ + update = False + if not self.safe_delete_mode: # This could just check self, refactor. + logger.warning(f"{nautobot_object} will be deleted as safe delete mode is not enabled.") + # This allows private class naming of nautobot objects to be ordered for delete() + # Example definition in adapter class var: _site = Location + self.diffsync.objects_to_delete[f"_{nautobot_object.__class__.__name__.lower()}"].append( + nautobot_object + ) # pylint: disable=protected-access + super().delete() + else: + if safe_delete_status: + safe_delete_status = Status.objects.get(name=safe_delete_status.capitalize()) + if hasattr(nautobot_object, "status"): + if not nautobot_object.status == safe_delete_status: + nautobot_object.status = safe_delete_status + logger.warning(f"{nautobot_object} has changed status to {safe_delete_status}.") + update = True + else: + # Not everything has a status. This may come in handy once more models are synced. + logger.warning(f"{nautobot_object} has no Status attribute.") + if hasattr(nautobot_object, "tags"): + ssot_safe_tag, _ = Tag.objects.get_or_create( + name="SSoT Safe Delete", + defaults={ + "description": "Safe Delete Mode tag to flag an object, but not delete from Nautobot.", + "color": ColorChoices.COLOR_RED, + }, + ) + object_tags = nautobot_object.tags.all() + # No exception raised for empty iterator, safe to do this any + if not any(obj_tag for obj_tag in object_tags if obj_tag.name == ssot_safe_tag.name): + nautobot_object.tags.add(ssot_safe_tag) + logger.warning(f"Tagging {nautobot_object} with `ssot-safe-delete`.") + update = True + if update: + tonb_nbutils.tag_object(nautobot_object=nautobot_object, custom_field="ssot-synced-from-ipfabric") + else: + logger.warning(f"{nautobot_object} has previously been tagged with `ssot-safe-delete`. Skipping...") + + return self + + +class Location(DiffSyncExtras): + """Location model.""" + + _modelname = "location" + _identifiers = ("name",) + _attributes = ("site_id", "status") + _children = {"device": "devices", "vlan": "vlans"} + + name: str + site_id: Optional[str] + status: str + devices: List["Device"] = list() # pylint: disable=use-list-literal + vlans: List["Vlan"] = list() # pylint: disable=use-list-literal + + @classmethod + def create(cls, diffsync, ids, attrs): + """Create Location in Nautobot.""" + tonb_nbutils.create_location(location_name=ids["name"], location_id=attrs["site_id"]) + return super().create(ids=ids, diffsync=diffsync, attrs=attrs) + + def delete(self) -> Optional["DiffSyncModel"]: + """Delete Location in Nautobot.""" + location = NautobotLocation.objects.get(name=self.name) + + self.safe_delete( + location, + SAFE_DELETE_LOCATION_STATUS, + ) + return super().delete() + + def update(self, attrs): + """Update Location Object in Nautobot.""" + location = NautobotLocation.objects.get(name=self.name) + if attrs.get("site_id"): + location.custom_field_data["ipfabric-site-id"] = attrs.get("site_id") + location.validated_save() + if attrs.get("status") == "Active": + safe_delete_tag, _ = Tag.objects.get_or_create(name="SSoT Safe Delete") + if not location.status == "Active": + location.status = Status.objects.get(name="Active") + device_tags = location.tags.filter(pk=safe_delete_tag.pk) + if device_tags.exists(): + location.tags.remove(safe_delete_tag) + tonb_nbutils.tag_object(nautobot_object=location, custom_field="ssot-synced-from-ipfabric") + return super().update(attrs) + + +class Device(DiffSyncExtras): + """Device model.""" + + _modelname = "device" + _identifiers = ("name",) + _attributes = ("location_name", "model", "vendor", "serial_number", "role", "status") + _children = {"interface": "interfaces"} + + name: str + location_name: Optional[str] + model: Optional[str] + vendor: Optional[str] + serial_number: Optional[str] + role: Optional[str] + status: Optional[str] + + mgmt_address: Optional[str] + + interfaces: List["Interface"] = list() # pylint: disable=use-list-literal + + @classmethod + def create(cls, diffsync, ids, attrs): + """Create Device in Nautobot under its parent location.""" + # Get DeviceType + device_type_filter = DeviceType.objects.filter(model=attrs["model"]) + if device_type_filter.exists(): + device_type_object = device_type_filter.first() + else: + device_type_object = tonb_nbutils.create_device_type_object( + device_type=attrs["model"], vendor_name=attrs["vendor"] + ) + # Get Role, update if missing cf and create otherwise + role_name = attrs.get("role", DEFAULT_DEVICE_ROLE) + device_role_filter = Role.objects.filter(name=role_name) + if device_role_filter.exists(): + device_role_object = device_role_filter.first() + device_role_object.cf["ipfabric_type"] = role_name + device_role_object.validated_save() + else: + device_role_object = tonb_nbutils.get_or_create_device_role_object( + role_name=role_name, role_color=DEFAULT_DEVICE_ROLE_COLOR + ) + # Get Status + device_status_filter = Status.objects.filter(name=DEFAULT_DEVICE_STATUS) + if device_status_filter.exists(): + device_status_object = device_status_filter.first() + else: + device_status_object = tonb_nbutils.create_status(DEFAULT_DEVICE_STATUS, DEFAULT_DEVICE_STATUS_COLOR) + # Get Location + location_object_filter = NautobotLocation.objects.filter(name=attrs["location_name"]) + if location_object_filter.exists(): + location = location_object_filter.first() + else: + location = tonb_nbutils.create_location(attrs["location_name"]) + + new_device, _ = NautobotDevice.objects.get_or_create( + name=ids["name"], + serial=attrs.get("serial_number", ""), + status=device_status_object, + device_type=device_type_object, + role=device_role_object, + location=location, + ) + try: + # Validated save happens inside of tag_objet + tonb_nbutils.tag_object(nautobot_object=new_device, custom_field="ssot-synced-from-ipfabric") + except ValidationError as error: + message = f"Unable to create device: {ids['name']}. A validation error occured. Enable debug for more information." + if diffsync.job.debug: + logger.debug(error) + logger.error(message) + + return super().create(ids=ids, diffsync=diffsync, attrs=attrs) + + def delete(self) -> Optional["DiffSyncModel"]: + """Delete device in Nautobot.""" + try: + device_object = NautobotDevice.objects.get(name=self.name) + self.safe_delete( + device_object, + SAFE_DELETE_DEVICE_STATUS, + ) + return super().delete() + except NautobotDevice.DoesNotExist: + logger.warning(f"Unable to match device by name, {self.name}") + + def update(self, attrs): + """Update devices in Nautobot based on Source.""" + try: + _device = NautobotDevice.objects.get(name=self.name) + if attrs.get("status") == "Active": + safe_delete_tag, _ = Tag.objects.get_or_create(name="SSoT Safe Delete") + if not _device.status == "Active": + _device.status = Status.objects.get(name="Active") + device_tags = _device.tags.filter(pk=safe_delete_tag.pk) + if device_tags.exists(): + _device.tags.remove(safe_delete_tag) + # TODO: If only the "model" is changing, the "vendor" is not available + if attrs.get("model"): + device_type_object = tonb_nbutils.create_device_type_object( + device_type=attrs["model"], vendor_name=attrs["vendor"] + ) + _device.type = device_type_object + if attrs.get("location_name"): + location = tonb_nbutils.create_location(attrs["location_name"]) + _device.location = location + if attrs.get("serial_number"): + _device.serial = attrs.get("serial_number") + if attrs.get("role"): + device_role_object = tonb_nbutils.get_or_create_device_role_object( + role_name=attrs.get("role", DEFAULT_DEVICE_ROLE), role_color=DEFAULT_DEVICE_ROLE_COLOR + ) + _device.device_role = device_role_object + tonb_nbutils.tag_object(nautobot_object=_device, custom_field="ssot-synced-from-ipfabric") + # Call the super().update() method to update the in-memory DiffSyncModel instance + return super().update(attrs) + except NautobotDevice.DoesNotExist: + logger.warning(f"Unable to match device by name, {self.name}") + + +class Interface(DiffSyncExtras): + """Interface model.""" + + _modelname = "interface" + _identifiers = ( + "name", + "device_name", + ) + _shortname = ("name",) + _attributes = ( + "description", + "enabled", + "mac_address", + "mtu", + "type", + "mgmt_only", + "ip_address", + "subnet_mask", + "ip_is_primary", + "status", + ) + + name: str + device_name: str + description: Optional[str] + enabled: Optional[bool] + mac_address: Optional[str] + mtu: Optional[int] + type: Optional[str] + mgmt_only: Optional[bool] + ip_address: Optional[str] + subnet_mask: Optional[str] + ip_is_primary: Optional[bool] + status: str + + @classmethod + def create(cls, diffsync, ids, attrs): + """Create interface in Nautobot under its parent device.""" + ssot_tag, _ = Tag.objects.get_or_create(name="SSoT Synced from IPFabric") + device_obj = NautobotDevice.objects.filter(Q(name=ids["device_name"]) & Q(tags__name=ssot_tag.name)).first() + + if not attrs.get("mac_address"): + attrs["mac_address"] = DEFAULT_INTERFACE_MAC + interface_obj = tonb_nbutils.create_interface( + device_obj=device_obj, + interface_details={**ids, **attrs}, + ) + ip_address = attrs["ip_address"] + if ip_address: + if interface_obj.ip_addresses.all().exists(): + interface_obj.ip_addresses.all().delete() + ip_address_obj = tonb_nbutils.create_ip( + ip_address=attrs["ip_address"], + subnet_mask=attrs["subnet_mask"], + status=attrs["status"], + object_pk=interface_obj, + ) + interface_obj.ip_addresses.add(ip_address_obj) + if attrs.get("ip_is_primary"): + if ip_address_obj.ip_version == 4: + device_obj.primary_ip4 = ip_address_obj + device_obj.save() + if ip_address_obj.ip_version == 6: + device_obj.primary_ip6 = ip_address_obj + device_obj.save() + interface_obj.save() + return super().create(ids=ids, diffsync=diffsync, attrs=attrs) + + def delete(self) -> Optional["DiffSyncModel"]: + """Delete Interface Object.""" + try: + ssot_tag, _ = Tag.objects.get_or_create(name="SSoT Synced from IPFabric") + device = NautobotDevice.objects.filter(Q(name=self.device_name) & Q(tags__name=ssot_tag.name)).first() + if not device: + return + interface = device.interfaces.get(name=self.name) + # Access the addr within an interface, change the status if necessary + if interface.ip_addresses.first(): + self.safe_delete(interface.ip_addresses.first(), SAFE_DELETE_IPADDRESS_STATUS) + # Then do the parent interface + # Attached interfaces do not have a status to update. + self.safe_delete( + interface, + ) + return super().delete() + except NautobotDevice.DoesNotExist: + logger.warning(f"Unable to match device by name, {self.name}") + + def update(self, attrs): # pylint: disable=too-many-branches + """Update Interface object in Nautobot.""" + try: + ssot_tag, _ = Tag.objects.get_or_create(name="SSoT Synced from IPFabric") + device = NautobotDevice.objects.filter(Q(name=self.device_name) & Q(tags__name=ssot_tag.name)).first() + interface = device.interfaces.get(name=self.name) + if attrs.get("description"): + interface.description = attrs["description"] + if attrs.get("enabled"): + interface.enabled = attrs["enabled"] + if attrs.get("mac_address"): + interface.mac_address = attrs["mac_address"] + if attrs.get("mtu"): + interface.mtu = attrs["mtu"] + if attrs.get("mode"): + interface.mode = attrs["mode"] + if attrs.get("lag"): + interface.lag = attrs["lag"] + if attrs.get("type"): + interface.type = attrs["type"] + if attrs.get("mgmt_only"): + interface.mgmt_only = attrs["mgmt_only"] + if attrs.get("ip_address"): + if interface.ip_addresses.all().exists(): + logger.info(f"Replacing IP from interface {interface} on {device.name}") + interface.ip_addresses.all().delete() + ip_address_obj = tonb_nbutils.create_ip( + ip_address=attrs.get("ip_address"), + subnet_mask=attrs.get("subnet_mask") if attrs.get("subnet_mask") else "255.255.255.255", + status="Active", + object_pk=interface, + ) + interface.ip_addresses.add(ip_address_obj) + if attrs.get("ip_is_primary"): + interface_obj = interface.ip_addresses.first() + if interface_obj: + if interface_obj.ip_version == 4: + device.primary_ip4 = interface_obj + device.save() + if interface_obj.ip_version == 6: + device.primary_ip6 = interface_obj + device.save() + interface.save() + tonb_nbutils.tag_object(nautobot_object=interface, custom_field="ssot-synced-from-ipfabric") + return super().update(attrs) + + except NautobotDevice.DoesNotExist: + logger.warning(f"Unable to match device by name, {self.name}") + + +class Vlan(DiffSyncExtras): + """VLAN model.""" + + _modelname = "vlan" + _identifiers = ("name", "location") + _shortname = ("name",) + _attributes = ("vid", "status", "description") + + name: str + vid: int + status: str + location: str + description: Optional[str] + vlan_pk: Optional[UUID] + + @classmethod + def create(cls, diffsync, ids, attrs): + """Create VLANs in Nautobot under the site.""" + status = attrs["status"].lower().capitalize() + location = NautobotLocation.objects.get(name=ids["location"]) + name = ids["name"] if ids["name"] else f"VLAN{attrs['vid']}" + description = attrs["description"] if attrs["description"] else None + if diffsync.job.debug: + logger.debug("Creating VLAN: %s description: %s", name, description) + tonb_nbutils.create_vlan( + vlan_name=name, + vlan_id=attrs["vid"], + vlan_status=status, + location_obj=location, + description=description, + ) + return super().create(ids=ids, diffsync=diffsync, attrs=attrs) + + def delete(self) -> Optional["DiffSyncModel"]: + """Delete.""" + vlan = VLAN.objects.get(name=self.name, pk=self.vlan_pk) + self.safe_delete( + vlan, + SAFE_DELETE_VLAN_STATUS, + ) + return super().delete() + + def update(self, attrs): + """Update VLAN object in Nautobot.""" + vlan = VLAN.objects.get(name=self.name, vid=self.vid, location=NautobotLocation.objects.get(name=self.location)) + + if attrs.get("status") == "Active": + safe_delete_tag, _ = Tag.objects.get_or_create(name="SSoT Safe Delete") + if not vlan.status == "Active": + vlan.status = Status.objects.get(name="Active") + device_tags = vlan.tags.filter(pk=safe_delete_tag.pk) + if device_tags.exists(): + vlan.tags.remove(safe_delete_tag) + if attrs.get("description"): + vlan.description = vlan.description + + tonb_nbutils.tag_object(nautobot_object=vlan, custom_field="ssot-synced-from-ipfabric") + + +Location.update_forward_refs() +Device.update_forward_refs() +Interface.update_forward_refs() +Vlan.update_forward_refs() diff --git a/nautobot_ssot/integrations/ipfabric/jobs.py b/nautobot_ssot/integrations/ipfabric/jobs.py new file mode 100644 index 000000000..b98b65060 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/jobs.py @@ -0,0 +1,313 @@ +# pylint: disable=keyword-arg-before-vararg +# pylint: disable=too-few-public-methods +# pylint: disable=too-many-locals +"""IP Fabric Data Target Job.""" +import uuid +from diffsync.exceptions import ObjectNotCreated +from django.conf import settings +from django.templatetags.static import static +from django.urls import reverse +from httpx import ConnectError +from ipfabric import IPFClient +from nautobot.dcim.models import Location +from nautobot.extras.jobs import BooleanVar, ScriptVariable, ChoiceVar +from nautobot.core.forms import DynamicModelChoiceField +from nautobot_ssot.jobs.base import DataMapping, DataSource + +from nautobot_ssot.integrations.ipfabric.diffsync.adapter_ipfabric import IPFabricDiffSync +from nautobot_ssot.integrations.ipfabric.diffsync.adapter_nautobot import NautobotDiffSync +from nautobot_ssot.integrations.ipfabric.diffsync.adapters_shared import DiffSyncModelAdapters +from nautobot_ssot.integrations.ipfabric.diffsync.diffsync_models import DiffSyncExtras +from nautobot_ssot.integrations.ipfabric import constants + +CONFIG = settings.PLUGINS_CONFIG.get("nautobot_ssot", {}) +IPFABRIC_HOST = CONFIG["ipfabric_host"] +IPFABRIC_API_TOKEN = CONFIG["ipfabric_api_token"] +IPFABRIC_SSL_VERIFY = CONFIG["ipfabric_ssl_verify"] +IPFABRIC_TIMEOUT = CONFIG["ipfabric_timeout"] +LAST = "$last" +PREV = "$prev" +LAST_LOCKED = "$lastLocked" + +name = "SSoT - IPFabric" # pylint: disable=invalid-name + + +def is_valid_uuid(identifier): + """Return true if the identifier it's a valid UUID.""" + try: + uuid.UUID(str(identifier)) + return True + except ValueError: + return False + + +def get_formatted_snapshots(client: IPFClient): + """Get all loaded snapshots and format them for display in choice menu. + + Returns: + dict: Snapshot objects as dict of tuples {snapshot_ref: (description, snapshot_id)} + """ + formatted_snapshots = {} + snapshot_refs = [] + if client: + for snapshot_ref, snapshot in client.loaded_snapshots.items(): + description = "" + if snapshot_ref in [LAST, PREV, LAST_LOCKED]: + description += f"{snapshot_ref}: " + snapshot_refs.append(snapshot_ref) + if snapshot.name: + description += snapshot.name + " - " + snapshot.end.strftime("%d-%b-%y %H:%M:%S") + else: + description += snapshot.end.strftime("%d-%b-%y %H:%M:%S") + " - " + snapshot.snapshot_id + formatted_snapshots[snapshot_ref] = (description, snapshot.snapshot_id) + for ref in snapshot_refs: + formatted_snapshots.pop(formatted_snapshots[ref][1], None) + + return formatted_snapshots + + +class OptionalObjectVar(ScriptVariable): + """Custom implementation of an Optional ObjectVar. + + An object primary key is returned and accessible in job kwargs. + """ + + kwargs = {} + form_field = DynamicModelChoiceField + + def __init__( + self, + model=None, + display_field="display", + query_params=None, + null_option=None, + *args, + **kwargs, + ): + """Init.""" + super().__init__(*args, **kwargs) + + if model is not None: + self.field_attrs["queryset"] = model.objects.all() + else: + raise TypeError("ObjectVar must specify a model") + + self.field_attrs.update( + { + "display_field": display_field, + "query_params": query_params, + "null_option": null_option, + } + ) + + +# pylint:disable=too-few-public-methods +class IpFabricDataSource(DataSource): + """Job syncing data from IP Fabric to Nautobot.""" + + client = None + snapshot = None + debug = BooleanVar(description="Enable for more verbose debug logging") + safe_delete_mode = BooleanVar( + description="Records are not deleted. Status fields are updated as necessary.", + default=True, + label="Safe Delete Mode", + ) + sync_ipfabric_tagged_only = BooleanVar( + default=True, + label="Sync Tagged Only", + description="Only sync objects that have the 'ssot-synced-from-ipfabric' tag.", + ) + location_filter = OptionalObjectVar( + description="Only sync Nautobot records belonging to a single Location. This does not filter IPFabric data.", + model=Location, + required=False, + ) + kwargs = {} + + class Meta: + """Metadata about this Job.""" + + name = "IPFabric ⟹ Nautobot" + data_source = "IP Fabric" + data_source_icon = static("nautobot_ssot_ipfabric/ipfabric.png") + description = "Sync data from IP Fabric into Nautobot." + field_order = ( + "debug", + "snapshot", + "safe_delete_mode", + "sync_ipfabric_tagged_only", + "dryrun", + ) + + @staticmethod + def _init_ipf_client(): + try: + return IPFClient( + base_url=IPFABRIC_HOST, + token=IPFABRIC_API_TOKEN, + verify=IPFABRIC_SSL_VERIFY, + timeout=IPFABRIC_TIMEOUT, + unloaded=False, + ) + except (RuntimeError, ConnectError) as error: + print(f"Got an error {error}") + return None + + @classmethod + def _get_vars(cls): + """Extend JobDataSource._get_vars to include some variables. + + This also initializes them. + """ + got_vars = super()._get_vars() + + if cls.client is None: + cls.client = cls._init_ipf_client() + else: + cls.client.update() + + formatted_snapshots = get_formatted_snapshots(cls.client) + if formatted_snapshots: + default_choice = formatted_snapshots["$last"][::-1] + else: + default_choice = "$last" + + cls.snapshot = ChoiceVar( + description="IPFabric snapshot to sync from. Defaults to $last", + default=default_choice, + choices=[(snapshot_id, snapshot_name) for snapshot_name, snapshot_id in formatted_snapshots.values()], + required=False, + ) + + if hasattr(cls, "snapshot"): + got_vars["snapshot"] = cls.snapshot + + return got_vars + + @classmethod + def data_mappings(cls): + """List describing the data mappings involved in this DataSource.""" + return ( + DataMapping("Device", None, "Device", reverse("dcim:device_list")), + DataMapping("Location", None, "Location", reverse("dcim:location_list")), + DataMapping("Interfaces", None, "Interfaces", reverse("dcim:interface_list")), + DataMapping("IP Addresses", None, "IP Addresses", reverse("ipam:ipaddress_list")), + DataMapping("VLANs", None, "VLANs", reverse("ipam:vlan_list")), + ) + + @classmethod + def config_information(cls): + """Dictionary describing the configuration of this DataSource.""" + return { + "IP Fabric host": CONFIG["ipfabric_host"], + "Nautobot Host URL": CONFIG.get("nautobot_host"), + "Default MAC Address": constants.DEFAULT_INTERFACE_MAC, + "Default Device Role": constants.DEFAULT_DEVICE_ROLE, + "Default Interface Type": constants.DEFAULT_INTERFACE_TYPE, + "Default Device Status": constants.DEFAULT_DEVICE_STATUS, + "Allow Duplicate Addresses": constants.ALLOW_DUPLICATE_ADDRESSES, + "Default MTU": constants.DEFAULT_INTERFACE_MTU, + "Safe Delete Device Status": constants.SAFE_DELETE_DEVICE_STATUS, + "Safe Delete Location Status": constants.SAFE_DELETE_LOCATION_STATUS, + "Safe Delete IPAddress Status": constants.SAFE_IPADDRESS_INTERFACES_STATUS, + "Safe Delete VLAN status": constants.SAFE_DELETE_VLAN_STATUS, + } + + # pylint: disable-next=too-many-arguments, arguments-differ + def run( + self, + dryrun, + memory_profiling, + debug, + snapshot=None, + safe_delete_mode=True, + sync_ipfabric_tagged_only=True, + location_filter=None, + *args, + **kwargs, + ): + """Run the job.""" + self.kwargs = { + "snapshot": snapshot, + "dryrun": dryrun, + "safe_delete_mode": safe_delete_mode, + "sync_ipfabric_tagged_only": sync_ipfabric_tagged_only, + "location_filter": location_filter, + "debug": debug, + } + + super().run(dryrun=dryrun, memory_profiling=memory_profiling, *args, **kwargs) + + def load_source_adapter(self): + """Not used.""" + + def load_target_adapter(self): + """Not used.""" + + def sync_data(self, *_args, **_kwargs): + """Sync a device data from IP Fabric into Nautobot.""" + if self.client is None: + self.client = self._init_ipf_client() + if self.client is None: + self.logger.error("IPFabric client is not ready. Check your config.") + return + + self.client.snapshot_id = self.kwargs["snapshot"] + dryrun = self.kwargs["dryrun"] + safe_mode = self.kwargs["safe_delete_mode"] + tagged_only = self.kwargs["sync_ipfabric_tagged_only"] + location_filter = self.kwargs["location_filter"] + debug_mode = self.kwargs["debug"] + + if location_filter: + location_filter_object = Location.objects.get(pk=location_filter) + else: + location_filter_object = None + options = f"`Snapshot_id`: {self.client.snapshot_id}.`Debug`: {debug_mode}, `Dry Run`: {dryrun}, `Safe Delete Mode`: {safe_mode}, `Sync Tagged Only`: {tagged_only}, `Location Filter`: {location_filter_object}" + self.logger.info(f"Starting job with the following options: {options}") + + ipfabric_source = IPFabricDiffSync(job=self, sync=self.sync, client=self.client) + self.logger.info("Loading current data from IP Fabric...") + ipfabric_source.load() + + # Set safe mode either way (Defaults to True) + DiffSyncModelAdapters.safe_delete_mode = safe_mode + DiffSyncExtras.safe_delete_mode = safe_mode + + dest = NautobotDiffSync( + job=self, + sync=self.sync, + sync_ipfabric_tagged_only=tagged_only, + location_filter=location_filter_object, + ) + + self.logger.info("Loading current data from Nautobot...") + dest.load() + self.logger.info("Calculating diffs...") + + diff = dest.diff_from(ipfabric_source) + # pylint: disable-next=logging-fstring-interpolation + if debug_mode: + self.logger.debug("Diff: %s", diff.dict()) + + self.sync.diff = diff.dict() + self.sync.save() + create = diff.summary().get("create") + update = diff.summary().get("update") + delete = diff.summary().get("delete") + no_change = diff.summary().get("no-change") + self.logger.info( + f"DiffSync Summary: Create: {create}, Update: {update}, Delete: {delete}, No Change: {no_change}" + ) + if not dryrun: + self.logger.info("Syncing from IP Fabric to Nautobot") + try: + dest.sync_from(ipfabric_source) + except ObjectNotCreated: + self.logger.debug("Unable to create object.", exc_info=True) + + self.logger.info("Sync complete.") + + +jobs = [IpFabricDataSource] diff --git a/nautobot_ssot/integrations/ipfabric/signals.py b/nautobot_ssot/integrations/ipfabric/signals.py new file mode 100644 index 000000000..3fadb9284 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/signals.py @@ -0,0 +1,81 @@ +# pylint: disable=duplicate-code +"""Signal handlers for IPFabric integration.""" + +from typing import List, Optional + +from nautobot.core.signals import nautobot_database_ready +from nautobot.extras.choices import CustomFieldTypeChoices +from nautobot.core.choices import ColorChoices + + +def register_signals(sender): + """Register signals for IPFabric integration.""" + nautobot_database_ready.connect(nautobot_database_ready_callback, sender=sender) + + +def create_custom_field(key: str, label: str, models: List, apps, cf_type: Optional[str] = "type_date"): + """Create custom field on a given model instance type. + + Args: + key (str): Natural key + label (str): Label description + models (List): List of Django Models + apps: Django Apps + cf_type: (str, optional): Type of Field. Supports 'type_text' or 'type_date'. Defaults to 'type_date'. + """ + ContentType = apps.get_model("contenttypes", "ContentType") # pylint:disable=invalid-name + CustomField = apps.get_model("extras", "CustomField") # pylint:disable=invalid-name + if cf_type == "type_date": + custom_field, _ = CustomField.objects.get_or_create( + key=key, + type=CustomFieldTypeChoices.TYPE_DATE, + label=label, + ) + else: + custom_field, _ = CustomField.objects.get_or_create( + key=key, + type=CustomFieldTypeChoices.TYPE_TEXT, + label=label, + ) + for model in models: + custom_field.content_types.add(ContentType.objects.get_for_model(model)) + custom_field.save() + + +def nautobot_database_ready_callback(sender, *, apps, **kwargs): # pylint: disable=unused-argument + """Callback function triggered by the nautobot_database_ready signal when the Nautobot database is fully ready.""" + # pylint: disable=invalid-name, too-many-locals + Device = apps.get_model("dcim", "Device") + DeviceType = apps.get_model("dcim", "DeviceType") + Role = apps.get_model("extras", "Role") + Interface = apps.get_model("dcim", "Interface") + IPAddress = apps.get_model("ipam", "IPAddress") + Manufacturer = apps.get_model("dcim", "Manufacturer") + Location = apps.get_model("dcim", "Location") + VLAN = apps.get_model("ipam", "VLAN") + Tag = apps.get_model("extras", "Tag") + ContentType = apps.get_model("contenttypes", "ContentType") + location_type = apps.get_model("dcim", "LocationType") + + Tag.objects.get_or_create( + name="SSoT Synced from IPFabric", + defaults={ + "description": "Object synced at some point from IPFabric to Nautobot", + "color": ColorChoices.COLOR_LIGHT_GREEN, + }, + ) + Tag.objects.get_or_create( + name="SSoT Safe Delete", + defaults={ + "description": "Safe Delete Mode tag to flag an object, but not delete from Nautobot.", + "color": ColorChoices.COLOR_RED, + }, + ) + loc_type, _ = location_type.objects.update_or_create(name="Site") + loc_type.content_types.add(ContentType.objects.get_for_model(Device)) + loc_type.content_types.add(ContentType.objects.get_for_model(apps.get_model("ipam", "Prefix"))) + loc_type.content_types.add(ContentType.objects.get_for_model(VLAN)) + synced_from_models = [Device, DeviceType, Interface, Manufacturer, Location, VLAN, Role, IPAddress] + create_custom_field("ssot-synced-from-ipfabric", "Last synced from IPFabric on", synced_from_models, apps=apps) + create_custom_field("ipfabric-site-id", "IPFabric Location ID", [Location], apps=apps, cf_type="type_text") + create_custom_field("ipfabric_type", "IPFabric Type", [Role], apps=apps, cf_type="type_text") diff --git a/nautobot_ssot/integrations/ipfabric/utilities/__init__.py b/nautobot_ssot/integrations/ipfabric/utilities/__init__.py new file mode 100644 index 000000000..5db34ba1f --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/utilities/__init__.py @@ -0,0 +1,25 @@ +"""Utilities.""" +from .nbutils import ( + get_or_create_device_role_object, + create_device_type_object, + create_interface, + create_ip, + create_manufacturer, + create_location, + create_status, + create_vlan, +) +from .test_utils import clean_slate, json_fixture + +__all__ = ( + "create_location", + "create_device_type_object", + "create_manufacturer", + "get_or_create_device_role_object", + "create_status", + "create_ip", + "create_interface", + "json_fixture", + "create_vlan", + "clean_slate", +) diff --git a/nautobot_ssot/integrations/ipfabric/utilities/nbutils.py b/nautobot_ssot/integrations/ipfabric/utilities/nbutils.py new file mode 100644 index 000000000..554f18539 --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/utilities/nbutils.py @@ -0,0 +1,256 @@ +# pylint: disable=duplicate-code +"""Utility functions for Nautobot ORM.""" +import datetime +from typing import Any, Optional + +from django.contrib.contenttypes.models import ContentType +from django.db import IntegrityError +from django.core.exceptions import ValidationError +from nautobot.dcim.models import ( + Device, + DeviceType, + Interface, + Manufacturer, + Location, + LocationType, +) +from nautobot.extras.choices import CustomFieldTypeChoices +from nautobot.extras.models import CustomField, Role, Tag +from nautobot.extras.models.statuses import Status +from nautobot.ipam.models import VLAN, IPAddress, Namespace, Prefix +from nautobot.ipam.choices import PrefixTypeChoices +from nautobot.core.choices import ColorChoices +from netutils.ip import netmask_to_cidr + + +def create_location(location_name, location_id=None): + """Creates a specified location in Nautobot. + + Args: + location_name (str): Name of the location. + location_id (str): ID of the location. + """ + location_obj, _ = Location.objects.get_or_create( + name=location_name, + location_type=LocationType.objects.get(name="Site"), + status=Status.objects.get(name="Active"), + ) + if location_id: + # Ensure custom field is available + custom_field_obj, _ = CustomField.objects.get_or_create( + type=CustomFieldTypeChoices.TYPE_TEXT, + label="ipfabric-site-id", + defaults={"label": "IPFabric Location ID"}, + ) + custom_field_obj.content_types.add(ContentType.objects.get_for_model(Location)) + location_obj.cf["ipfabric-site-id"] = location_id + location_obj.validated_save() + tag_object(nautobot_object=location_obj, custom_field="ssot-synced-from-ipfabric") + return location_obj + + +def create_manufacturer(vendor_name): + """Create specified manufacturer in Nautobot.""" + mf_name, _ = Manufacturer.objects.get_or_create(name=vendor_name) + tag_object(nautobot_object=mf_name, custom_field="ssot-synced-from-ipfabric") + return mf_name + + +def create_device_type_object(device_type, vendor_name): + """Create a specified device type in Nautobot. + + Args: + device_type (str): Device model gathered from DiffSync model. + vendor_name (str): Vendor Name + """ + mf_name = create_manufacturer(vendor_name) + device_type_obj, _ = DeviceType.objects.get_or_create(manufacturer=mf_name, model=device_type) + tag_object(nautobot_object=device_type_obj, custom_field="ssot-synced-from-ipfabric") + return device_type_obj + + +def get_or_create_device_role_object(role_name, role_color): + """Create specified device role in Nautobot. + + Args: + role_name (str): Role name. + role_color (str): Role color. + """ + # adds custom field to map custom role names to ipfabric type names + try: + role_obj = Role.objects.get(_custom_field_data__ipfabric_type=role_name) + except Role.DoesNotExist: + role_obj = Role.objects.create(name=role_name, color=role_color) + role_obj.cf["ipfabric_type"] = role_name + role_obj.validated_save() + role_obj.content_types.set([ContentType.objects.get_for_model(Device)]) + tag_object(nautobot_object=role_obj, custom_field="ssot-synced-from-ipfabric") + return role_obj + + +def create_status(status_name, status_color, description="", app_label="dcim", model="device"): + """Verify status object exists in Nautobot. If not, creates specified status. Defaults to dcim | device. + + Args: + status_name (str): Status name. + status_color (str): Status color. + description (str): Description + app_label (str): App Label ("DCIM") + model (str): Django Model ("DEVICE") + """ + try: + status_obj = Status.objects.get(name=status_name) + except Status.DoesNotExist: + content_type = ContentType.objects.get(app_label=app_label, model=model) + status_obj = Status.objects.create( + name=status_name, + color=status_color, + description=description, + ) + status_obj.content_types.set([content_type]) + return status_obj + + +def create_ip(ip_address, subnet_mask, status="Active", object_pk=None): + """Verify ip address exists in Nautobot. If not, creates specified ip. + + Utility behavior is manipulated by `settings` if duplicate ip's are allowed. + + Args: + ip_address (str): IP address. + subnet_mask (str): Subnet mask used for IP Address. + status (str): Status to assign to IP Address. + object_pk: Object primary key + """ + status_obj = Status.objects.get_for_model(IPAddress).get(name=status) + namespace_obj = Namespace.objects.get(name="Global") + cidr = netmask_to_cidr(subnet_mask) + try: + ip_obj, _ = IPAddress.objects.get_or_create(address=f"{ip_address}/{cidr}", status=status_obj) + except ValidationError: + parent, _ = Prefix.objects.get_or_create( + network="0.0.0.0", # nosec B104 + prefix_length=0, + type=PrefixTypeChoices.TYPE_NETWORK, + status=Status.objects.get_for_model(Prefix).get(name="Active"), + namespace=namespace_obj, + ) + ip_obj, _ = IPAddress.objects.get_or_create(address=f"{ip_address}/{cidr}", status=status_obj, parent=parent) + + if object_pk: + ip_obj.assigned_object_id = object_pk.pk + ip_obj.assigned_object_type = ContentType.objects.get_for_model(type(object_pk)) + # Tag Interface (object_pk) + tag_object(nautobot_object=object_pk, custom_field="ssot-synced-from-ipfabric") + + # Tag IP Addr + tag_object(nautobot_object=ip_obj, custom_field="ssot-synced-from-ipfabric") + return ip_obj + + +def create_interface(device_obj, interface_details): + """Verify interface exists on specified device. If not, creates interface. + + Args: + device_obj (Device): Device object to check interface against. + interface_details (dict): interface details. + """ + interface_fields = ( + "name", + "description", + "enabled", + "mac_address", + "mtu", + "type", + "mgmt_only", + "status", + ) + fields = {k: v for k, v in interface_details.items() if k in interface_fields and v} + try: + fields["status"] = Status.objects.get_for_model(Interface).get(name=fields.get(fields["status"], "Active")) + interface_obj, _ = device_obj.interfaces.get_or_create(**fields) + except IntegrityError: + interface_obj, _ = device_obj.interfaces.get_or_create(name=fields["name"]) + interface_obj.description = fields.get("description", "") + interface_obj.enabled = fields.get("enabled") + interface_obj.mac_address = fields.get("mac_address") + interface_obj.mtu = fields.get("mtu") + interface_obj.type = fields.get("type") + interface_obj.mgmt_only = fields.get("mgmt_only", False) + interface_obj.status = Status.objects.get_for_model(Interface).get(name=fields.get("status", "Active")) + interface_obj.validated_save() + tag_object(nautobot_object=interface_obj, custom_field="ssot-synced-from-ipfabric") + return interface_obj + + +def create_vlan(vlan_name: str, vlan_id: int, vlan_status: str, location_obj: Location, description: str): + """Creates or obtains VLAN object. + + Args: + vlan_name (str): VLAN Name + vlan_id (int): VLAN ID + vlan_status (str): VLAN Status + location_obj (Location): Location Django Model + description (str): VLAN Description + + Returns: + (VLAN): Returns created or obtained VLAN object. + """ + vlan_obj, _ = location_obj.vlans.get_or_create( + name=vlan_name, vid=vlan_id, status=Status.objects.get(name=vlan_status), description=description + ) + tag_object(nautobot_object=vlan_obj, custom_field="ssot-synced-from-ipfabric") + return vlan_obj + + +def tag_object(nautobot_object: Any, custom_field: str, tag_name: Optional[str] = "SSoT Synced from IPFabric"): + """Apply the given tag and custom field to the identified object. + + Args: + nautobot_object (Any): Nautobot ORM Object + custom_field (str): Name of custom field to update + tag_name (Optional[str], optional): Tag name. Defaults to "SSoT Synced From IPFabric". + """ + if tag_name == "SSoT Synced from IPFabric": + tag, _ = Tag.objects.get_or_create( + name="SSoT Synced from IPFabric", + defaults={ + "description": "Object synced at some point from IPFabric to Nautobot", + "color": ColorChoices.COLOR_LIGHT_GREEN, + }, + ) + else: + tag, _ = Tag.objects.get_or_create(name=tag_name) + + today = datetime.date.today().isoformat() + + def _tag_object(nautobot_object): + """Apply custom field and tag to object, if applicable.""" + if hasattr(nautobot_object, "tags"): + nautobot_object.tags.add(tag) + if hasattr(nautobot_object, "cf"): + # Ensure that the "ssot-synced-from-ipfabric" custom field is present + if not any(cfield for cfield in CustomField.objects.all() if cfield.key == "ssot-synced-from-ipfabric"): + custom_field_obj, _ = CustomField.objects.get( + key="ssot-synced-from-ipfabric", + ) + synced_from_models = [ + Device, + DeviceType, + Interface, + Manufacturer, + Location, + VLAN, + Role, + IPAddress, + ] + for model in synced_from_models: + custom_field_obj.content_types.add(ContentType.objects.get_for_model(model)) + custom_field_obj.validated_save() + + # Update custom field date stamp + nautobot_object.cf[custom_field] = today + nautobot_object.validated_save() + + _tag_object(nautobot_object) + # Ensure proper save diff --git a/nautobot_ssot/integrations/ipfabric/utilities/test_utils.py b/nautobot_ssot/integrations/ipfabric/utilities/test_utils.py new file mode 100644 index 000000000..fb672dbfc --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/utilities/test_utils.py @@ -0,0 +1,21 @@ +"""Test Utils.""" +import json + +from nautobot.dcim.models import Device, Location +from nautobot.ipam.models import VLAN + + +def json_fixture(json_file_path): + """Load and return JSON Fixture.""" + with open(json_file_path, "r", encoding="utf-8") as file: + return json.load(file) + + +def clean_slate(): + """Delete all objects synced. + + Use this with caution. Never use in production env. + """ + VLAN.objects.all().delete() + Device.objects.all().delete() + Location.objects.all().delete() diff --git a/nautobot_ssot/integrations/ipfabric/workers.py b/nautobot_ssot/integrations/ipfabric/workers.py new file mode 100644 index 000000000..dad7cf6ea --- /dev/null +++ b/nautobot_ssot/integrations/ipfabric/workers.py @@ -0,0 +1,145 @@ +# Disable dispatcher from chatops unused. # pylint: disable=unused-argument +"""Chat Ops Worker.""" +import uuid + +from django.conf import settings +from django.contrib.contenttypes.models import ContentType +from nautobot.core.settings_funcs import is_truthy +from nautobot.extras.models import JobResult + +# pylint: disable-next=import-error +from nautobot_chatops.choices import CommandStatusChoices + +# pylint: disable-next=import-error +from nautobot_chatops.dispatchers import Dispatcher + +# pylint: disable-next=import-error +from nautobot_chatops.workers import handle_subcommands, subcommand_of + +from nautobot_ssot.integrations.ipfabric.jobs import IpFabricDataSource + +# from nautobot.dcim.models import Site + +CONFIG = settings.PLUGINS_CONFIG.get("nautobot_ssot", {}) +NAUTOBOT_HOST = CONFIG.get("nautobot_host") + +BASE_CMD = "ipfabric" +IPFABRIC_LOGO_PATH = "nautobot_ssot_ipfabric/ipfabric_logo.png" +IPFABRIC_LOGO_ALT = "IPFabric Logo" + + +def prompt_for_bool(dispatcher: Dispatcher, action_id: str, help_text: str): + """Prompt the user to select a True or False choice.""" + choices = [("Yes", "True"), ("No", "False")] + dispatcher.prompt_from_menu(action_id, help_text, choices, default=("Yes", "True")) + return False + + +# def prompt_for_site(dispatcher: Dispatcher, action_id: str, help_text: str, sites=None, offset=0): +# """Prompt the user to select a valid site from a drop-down menu.""" +# if sites is None: +# sites = Site.objects.all().order_by("name") +# if not sites: +# dispatcher.send_error("No sites were found") +# return (CommandStatusChoices.STATUS_FAILED, "No sites found") +# choices = [(f"{site.name}: {site.name}", site.name) for site in sites] +# return dispatcher.prompt_from_menu(action_id, help_text, choices, offset=offset) + + +def ipfabric_logo(dispatcher): + """Construct an image_element containing the locally hosted IP Fabric logo.""" + return dispatcher.image_element(dispatcher.static_url(IPFABRIC_LOGO_PATH), alt_text=IPFABRIC_LOGO_ALT) + + +def ipfabric(subcommand, **kwargs): + """Interact with ipfabric plugin.""" + return handle_subcommands("ipfabric", subcommand, **kwargs) + + +@subcommand_of("ipfabric") +def ssot_sync_to_nautobot( + dispatcher, + dry_run=None, + safe_delete_mode=None, + sync_ipfabric_tagged_only=None, +): + """Start an SSoT sync from IPFabric to Nautobot.""" + if dry_run is None: + prompt_for_bool(dispatcher, f"{BASE_CMD} ssot-sync-to-nautobot", "Do you want to run a `Dry Run`?") + return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + if safe_delete_mode is None: + prompt_for_bool( + dispatcher, f"{BASE_CMD} ssot-sync-to-nautobot {dry_run}", "Do you want to run in `Safe Delete Mode`?" + ) + return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + if sync_ipfabric_tagged_only is None: + prompt_for_bool( + dispatcher, + f"{BASE_CMD} ssot-sync-to-nautobot {dry_run} {safe_delete_mode}", + "Do you want to sync against `ssot-tagged-from-ipfabric` tagged objects only?", + ) + return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + # if location_filter is None: + # prompt_for_site( + # dispatcher, + # f"{BASE_CMD} ssot-sync-to-nautobot {dry_run} {safe_delete_mode} {sync_ipfabric_tagged_only}", + # "Select a Site to use as an optional filter?", + # ) + # return (CommandStatusChoices.STATUS_SUCCEEDED, "Success") + + # Implement filter in future release + location_filter = False + + sync_job = IpFabricDataSource() + + sync_job.job_result = JobResult( + name=sync_job.class_path, + obj_type=ContentType.objects.get( + app_label="extras", + model="job", + ), + job_id=uuid.uuid4(), + ) + sync_job.job_result.validated_save() + + dispatcher.send_markdown( + f"Stand by {dispatcher.user_mention()}, I'm running your sync with options set to `Dry Run`: {dry_run}, `Safe Delete Mode`: {safe_delete_mode}. `Sync Tagged Only`: {sync_ipfabric_tagged_only}", + ephemeral=True, + ) + + sync_job.run( + dryrun=is_truthy(dry_run), + memory_profiling=False, + safe_delete_mode=is_truthy(safe_delete_mode), + sync_ipfabric_tagged_only=is_truthy(sync_ipfabric_tagged_only), + location_filter=location_filter, + debug=False, + ) + sync_job.job_result.validated_save() + + blocks = [ + *dispatcher.command_response_header( + "ipfabric", + "ssot-sync-to-nautobot", + [ + ("Dry Run", str(dry_run)), + ("Safe Delete Mode", str(safe_delete_mode)), + ("Sync IPFabric Tagged Only", str(sync_ipfabric_tagged_only)), + ], + "sync job", + ipfabric_logo(dispatcher), + ), + ] + dispatcher.send_blocks(blocks) + if sync_job.job_result.status == "completed": + dispatcher.send_markdown( + f"Sync completed succesfully. Here is the link to your job: {NAUTOBOT_HOST}{sync_job.sync.get_absolute_url()}." + ) + else: + dispatcher.send_warning( + f"Sync failed. Here is the link to your job: {NAUTOBOT_HOST}{sync_job.sync.get_absolute_url()}" + ) + return CommandStatusChoices.STATUS_SUCCEEDED diff --git a/nautobot_ssot/static/nautobot_ssot_ipfabric/ipfabric.png b/nautobot_ssot/static/nautobot_ssot_ipfabric/ipfabric.png new file mode 100644 index 000000000..5d1073934 Binary files /dev/null and b/nautobot_ssot/static/nautobot_ssot_ipfabric/ipfabric.png differ diff --git a/nautobot_ssot/static/nautobot_ssot_ipfabric/ipfabric_logo.png b/nautobot_ssot/static/nautobot_ssot_ipfabric/ipfabric_logo.png new file mode 100644 index 000000000..7c83b27aa Binary files /dev/null and b/nautobot_ssot/static/nautobot_ssot_ipfabric/ipfabric_logo.png differ diff --git a/nautobot_ssot/tests/ipfabric/__init__.py b/nautobot_ssot/tests/ipfabric/__init__.py new file mode 100644 index 000000000..033383d79 --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/__init__.py @@ -0,0 +1 @@ +"""Unit tests for nautobot_ssot.integrations.ipfabric plugin.""" diff --git a/nautobot_ssot/tests/ipfabric/fixtures/__init__.py b/nautobot_ssot/tests/ipfabric/fixtures/__init__.py new file mode 100644 index 000000000..13092a6d2 --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/fixtures/__init__.py @@ -0,0 +1,17 @@ +"""Fixtures. + +In your test file, simply import: +``` +from nautobot_ssot.integrations.ipfabric.utilities import json_fixture +from nautobot_ssot.integrations.ipfabric.tests.fixtures import real_path +``` +Then you can simply load fixtures that you have added to the fixtures directory +and assign them to your mocks by using the json_fixture utility: + +json_fixture(f"{FIXTURES}/get_projects.json") + +This will return a loaded json object. +""" +import os + +real_path = os.path.dirname(os.path.realpath(__file__)) diff --git a/nautobot_ssot/tests/ipfabric/fixtures/get_device_inventory.json b/nautobot_ssot/tests/ipfabric/fixtures/get_device_inventory.json new file mode 100644 index 000000000..64da0491a --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/fixtures/get_device_inventory.json @@ -0,0 +1,68 @@ +[ + { + "hostname": "nyc-spine-02", + "siteName": "NYC-SPINE-02", + "vendor": "arista", + "platform": "veos", + "model": "CVX", + "memoryUtilization": 40.11, + "version": "4.22.4M", + "sn": "5254.00d3.a91d", + "loginIp": "172.18.0.22" + }, + { + "hostname": "jcy-spine-01.infra.ntc.com", + "siteName": "JCY-SPINE-01.INFRA.NTC.COM_1", + "vendor": "cisco", + "platform": "nx9000", + "model": "N9K-C9300v", + "memoryUtilization": 58.22, + "version": "9.3(3)", + "sn": "a000a04", + "loginIp": "172.18.0.9" + }, + { + "hostname": "jcy-rtr-01", + "siteName": "JCY-RTR-01_1", + "vendor": "cisco", + "platform": "csr1000", + "model": "CSR1000V", + "memoryUtilization": 11.97, + "version": "17.1.1", + "sn": "a000a01", + "loginIp": "172.18.0.3" + }, + { + "hostname": "nyc-rtr-01", + "siteName": "NYC-RTR-01", + "vendor": "juniper", + "platform": "vmx", + "model": "", + "memoryUtilization": 15, + "version": "18.2R1.9", + "sn": "VM60D5EE2211", + "loginIp": "172.18.0.14" + }, + { + "hostname": "nyc-leaf-01", + "siteName": "NYC-LEAF-01", + "vendor": "arista", + "platform": "veos", + "model": "CVX", + "memoryUtilization": 39.6, + "version": "4.22.4M", + "sn": "5254.0029.fbf2", + "loginIp": "172.18.0.25" + }, + { + "hostname": "jcy-rtr-02", + "siteName": "JCY-RTR-02_1", + "vendor": "cisco", + "platform": "csr1000", + "model": "CSR1000V", + "memoryUtilization": 11.97, + "version": "17.1.1", + "sn": "a000a02", + "loginIp": "172.18.0.5" + } +] \ No newline at end of file diff --git a/nautobot_ssot/tests/ipfabric/fixtures/get_interface_inventory.json b/nautobot_ssot/tests/ipfabric/fixtures/get_interface_inventory.json new file mode 100644 index 000000000..44fae79e7 --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/fixtures/get_interface_inventory.json @@ -0,0 +1,50 @@ +[ + { + "id":"19941192", + "hostname":"nyc-rtr-01", + "intName":"ipip", + "dscr":null, + "mac":null, + "duplex":null, + "speed":"unlimited", + "media":null, + "mtu":null, + "primaryIp":null + }, + { + "id":"19952051", + "hostname":"nyc-leaf-01", + "intName":"Et15", + "dscr":null, + "mac":"5254.009d.f816", + "duplex":"full", + "speed":"unconfigured", + "media":null, + "mtu":9214, + "primaryIp":null + }, + { + "id":"19957897", + "hostname":"jcy-rtr-02", + "intName":"Gi4", + "dscr":null, + "mac":"5254.0090.4b0a", + "duplex":"full", + "speed":1000000000, + "media":"Virtual", + "mtu":1500, + "primaryIp":"10.10.0.10" + }, + { + "id":"19941192", + "hostname":"nyc-rtr-01", + "intName":"eth1", + "dscr":null, + "mac":null, + "duplex":null, + "speed":"unlimited", + "media":null, + "mtu":null, + "primaryIp":null + } +] \ No newline at end of file diff --git a/nautobot_ssot/tests/ipfabric/fixtures/get_sites.json b/nautobot_ssot/tests/ipfabric/fixtures/get_sites.json new file mode 100644 index 000000000..dffb0c3fc --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/fixtures/get_sites.json @@ -0,0 +1,80 @@ +[ + { + "id": "55467784", + "siteName": "JCY-RTR-01_1", + "siteKey": "55467784", + "devicesCount": 1, + "usersCount": 0, + "stpDCount": 0, + "switchesCount": 0, + "vlanCount": 0, + "rDCount": 1, + "routersCount": 1, + "networksCount": 5 + }, + { + "id": "55467890", + "siteName": "JCY-RTR-02_1", + "siteKey": "55467890", + "devicesCount": 1, + "usersCount": 0, + "stpDCount": 0, + "switchesCount": 0, + "vlanCount": 0, + "rDCount": 1, + "routersCount": 1, + "networksCount": 5 + }, + { + "id": "55467964", + "siteName": "JCY-SPINE-01.INFRA.NTC.COM_1", + "siteKey": "55467964", + "devicesCount": 1, + "usersCount": 3, + "stpDCount": 0, + "switchesCount": 1, + "vlanCount": 11, + "rDCount": 1, + "routersCount": 1, + "networksCount": 7 + }, + { + "id": "55467605", + "siteName": "NYC-LEAF-01", + "siteKey": "55467605", + "devicesCount": 1, + "usersCount": 0, + "stpDCount": 0, + "switchesCount": 1, + "vlanCount": 1, + "rDCount": 1, + "routersCount": 1, + "networksCount": 4 + }, + { + "id": "55468036", + "siteName": "NYC-RTR-01", + "siteKey": "55468036", + "devicesCount": 1, + "usersCount": 0, + "stpDCount": 0, + "switchesCount": 0, + "vlanCount": 0, + "rDCount": 1, + "routersCount": 1, + "networksCount": 7 + }, + { + "id": "55467678", + "siteName": "NYC-SPINE-02", + "siteKey": "55467678", + "devicesCount": 1, + "usersCount": 0, + "stpDCount": 0, + "switchesCount": 1, + "vlanCount": 1, + "rDCount": 1, + "routersCount": 1, + "networksCount": 5 + } +] \ No newline at end of file diff --git a/nautobot_ssot/tests/ipfabric/fixtures/get_vlans.json b/nautobot_ssot/tests/ipfabric/fixtures/get_vlans.json new file mode 100644 index 000000000..806fdca6e --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/fixtures/get_vlans.json @@ -0,0 +1,80 @@ +[ + { + "siteName":"NYC-LEAF-01", + "vlanName":"default", + "vlanId":1, + "dscr":"None" + }, + { + "siteName":"NYC-SPINE-02", + "vlanName":"default", + "vlanId":1, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"default", + "vlanId":1, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"autovlan", + "vlanId":101, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"vlan109", + "vlanId":109, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"webvlan0001", + "vlanId":194, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"vlan202", + "vlanId":202, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"vlan203", + "vlanId":203, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"vlan204", + "vlanId":204, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"vlan205", + "vlanId":205, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"vlan280", + "vlanId":280, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"appvlan0004", + "vlanId":292, + "dscr":"None" + }, + { + "siteName":"JCY-SPINE-01.INFRA.NTC.COM_1", + "vlanName":"dbvlan0001", + "vlanId":393, + "dscr":"None" + } +] \ No newline at end of file diff --git a/nautobot_ssot/tests/ipfabric/test_ipfabric_adapter.py b/nautobot_ssot/tests/ipfabric/test_ipfabric_adapter.py new file mode 100644 index 000000000..18d2d5b28 --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/test_ipfabric_adapter.py @@ -0,0 +1,85 @@ +"""Unit tests for the IPFabric DiffSync adapter class.""" +import json +from unittest.mock import MagicMock + +from django.test import TestCase +from nautobot.extras.models import JobResult + +from nautobot_ssot.integrations.ipfabric.diffsync.adapter_ipfabric import IPFabricDiffSync +from nautobot_ssot.integrations.ipfabric.jobs import IpFabricDataSource + + +def load_json(path): + """Load a json file.""" + with open(path, encoding="utf-8") as file: + return json.loads(file.read()) + + +SITE_FIXTURE = load_json("./nautobot_ssot/tests/ipfabric/fixtures/get_sites.json") +DEVICE_INVENTORY_FIXTURE = load_json("./nautobot_ssot/tests/ipfabric/fixtures/get_device_inventory.json") +VLAN_FIXTURE = load_json("./nautobot_ssot/tests/ipfabric/fixtures/get_vlans.json") +INTERFACE_FIXTURE = load_json("./nautobot_ssot/tests/ipfabric/fixtures/get_interface_inventory.json") + + +class IPFabricDiffSyncTestCase(TestCase): + """Test the IPFabricDiffSync adapter class.""" + + def test_data_loading(self): + """Test the load() function.""" + + # Create a mock client + ipfabric_client = MagicMock() + ipfabric_client.inventory.sites.all.return_value = SITE_FIXTURE + ipfabric_client.inventory.devices.all.return_value = DEVICE_INVENTORY_FIXTURE + ipfabric_client.fetch_all = MagicMock( + side_effect=(lambda x: VLAN_FIXTURE if x == "tables/vlan/site-summary" else "") + ) + ipfabric_client.inventory.interfaces.all.return_value = INTERFACE_FIXTURE + + job = IpFabricDataSource() + job.job_result = JobResult.objects.create(name=job.class_path, task_name="fake task", worker="default") + ipfabric = IPFabricDiffSync(job=job, sync=None, client=ipfabric_client) + ipfabric.load() + self.assertEqual( + {site["siteName"] for site in SITE_FIXTURE}, + {site.get_unique_id() for site in ipfabric.get_all("location")}, + ) + self.assertEqual( + {dev["hostname"] for dev in DEVICE_INVENTORY_FIXTURE}, + {dev.get_unique_id() for dev in ipfabric.get_all("device")}, + ) + self.assertEqual( + {f"{vlan['vlanName']}__{vlan['siteName']}" for vlan in VLAN_FIXTURE}, + {vlan.get_unique_id() for vlan in ipfabric.get_all("vlan")}, + ) + + # Assert each site has a device tied to it. + for site in ipfabric.get_all("location"): + self.assertEqual(len(site.devices), 1, f"{site} does not have the expected single device tied to it.") + self.assertTrue(hasattr(site, "vlans")) + + # Assert each device has the necessary attributes + for device in ipfabric.get_all("device"): + self.assertTrue(hasattr(device, "location_name")) + self.assertTrue(hasattr(device, "model")) + self.assertTrue(hasattr(device, "vendor")) + self.assertTrue(hasattr(device, "serial_number")) + self.assertTrue(hasattr(device, "interfaces")) + + # Assert each vlan has the necessary attributes + for vlan in ipfabric.get_all("vlan"): + self.assertTrue(hasattr(vlan, "name")) + self.assertTrue(hasattr(vlan, "vid")) + self.assertTrue(hasattr(vlan, "status")) + self.assertTrue(hasattr(vlan, "location")) + self.assertTrue(hasattr(vlan, "description")) + + # Assert each interface has the necessary attributes + for interface in ipfabric.get_all("interface"): + self.assertTrue(hasattr(interface, "name")) + self.assertTrue(hasattr(interface, "device_name")) + self.assertTrue(hasattr(interface, "mac_address")) + self.assertTrue(hasattr(interface, "mtu")) + self.assertTrue(hasattr(interface, "ip_address")) + self.assertTrue(hasattr(interface, "subnet_mask")) + self.assertTrue(hasattr(interface, "type")) diff --git a/nautobot_ssot/tests/ipfabric/test_jobs.py b/nautobot_ssot/tests/ipfabric/test_jobs.py new file mode 100644 index 000000000..ac355629d --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/test_jobs.py @@ -0,0 +1,72 @@ +"""Test IPFabric Jobs.""" +from copy import deepcopy + +from django.conf import settings +from django.test import TestCase +from django.urls import reverse + +from nautobot_ssot.integrations.ipfabric import jobs + +CONFIG = settings.PLUGINS_CONFIG.get("nautobot_ssot", {}) +BACKUP_CONFIG = deepcopy(CONFIG) + + +class IPFabricJobTest(TestCase): + """Test the IPFabric job.""" + + def test_metadata(self): + """Verify correctness of the Job Meta attributes.""" + self.assertEqual("IPFabric ⟹ Nautobot", jobs.IpFabricDataSource.name) + self.assertEqual("IPFabric ⟹ Nautobot", jobs.IpFabricDataSource.Meta.name) + self.assertEqual("IP Fabric", jobs.IpFabricDataSource.Meta.data_source) + self.assertEqual("Sync data from IP Fabric into Nautobot.", jobs.IpFabricDataSource.Meta.description) + + def test_data_mapping(self): + """Verify correctness of the data_mappings() API.""" + mappings = jobs.IpFabricDataSource.data_mappings() + + self.assertEqual("Device", mappings[0].source_name) + self.assertIsNone(mappings[0].source_url) + self.assertEqual("Device", mappings[0].target_name) + self.assertEqual(reverse("dcim:device_list"), mappings[0].target_url) + + self.assertEqual("Location", mappings[1].source_name) + self.assertIsNone(mappings[1].source_url) + self.assertEqual("Location", mappings[1].target_name) + self.assertEqual(reverse("dcim:location_list"), mappings[1].target_url) + + self.assertEqual("Interfaces", mappings[2].source_name) + self.assertIsNone(mappings[2].source_url) + self.assertEqual("Interfaces", mappings[2].target_name) + self.assertEqual(reverse("dcim:interface_list"), mappings[2].target_url) + + self.assertEqual("IP Addresses", mappings[3].source_name) + self.assertIsNone(mappings[3].source_url) + self.assertEqual("IP Addresses", mappings[3].target_name) + self.assertEqual(reverse("ipam:ipaddress_list"), mappings[3].target_url) + + self.assertEqual("VLANs", mappings[4].source_name) + self.assertIsNone(mappings[4].source_url) + self.assertEqual("VLANs", mappings[4].target_name) + self.assertEqual(reverse("ipam:vlan_list"), mappings[4].target_url) + + # @override_settings( + # PLUGINS_CONFIG={ + # "nautobot_ssot": { + # "IPFABRIC_HOST": "https://ipfabric.networktocode.com", + # "IPFABRIC_API_TOKEN": "1234", + # } + # } + # ) + # def test_config_information(self): + # """Verify the config_information() API.""" + # CONFIG["ipfabric_host"] = "https://ipfabric.networktocode.com" + # config_information = jobs.IpFabricDataSource.config_information() + # self.assertContains( + # config_information, + # { + # "IP Fabric host": "https://ipfabric.networktocode.com", + # }, + # ) + # # CLEANUP + # CONFIG["ipfabric_host"] = BACKUP_CONFIG["ipfabric_host"] diff --git a/nautobot_ssot/tests/ipfabric/test_nautobot_adapter.py b/nautobot_ssot/tests/ipfabric/test_nautobot_adapter.py new file mode 100644 index 000000000..15963bf79 --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/test_nautobot_adapter.py @@ -0,0 +1,79 @@ +# """Unit tests for the IPFabric DiffSync adapter class.""" + +# import uuid + +# from django.contrib.contenttypes.models import ContentType +# from django.test import TestCase +# from nautobot.dcim.models import Device, DeviceRole, DeviceType, Manufacturer, Site +# from nautobot.extras.models import Job, JobResult, Status +# from nautobot.ipam.models import VLAN + +# from nautobot_ssot_ipfabric.diffsync.adapter_nautobot import NautobotDiffSync +# from nautobot_ssot_ipfabric.jobs import IpFabricDataSource + + +# class IPFabricDiffSyncTestCase(TestCase): +# """Test the NautobotDiffSync adapter class.""" + +# def setUp(self): +# """Create Nautobot objects to load and test with.""" +# status_active = Status.objects.get(slug="active") + +# site_1 = Site.objects.create(name="Site 1", slug="site-1", status=status_active) +# site_2 = Site.objects.create(name="Site 2", slug="site-2", status=status_active) + +# manufacturer = Manufacturer.objects.create(name="Cisco", slug="cisco") +# device_type = DeviceType.objects.create(manufacturer=manufacturer, model="CSR 1000v", slug="csr1000v") +# device_role = DeviceRole.objects.create(name="Router", slug="router") + +# Device.objects.create( +# name="csr1", device_type=device_type, device_role=device_role, site=site_1, status=status_active +# ) +# Device.objects.create( +# name="csr2", device_type=device_type, device_role=device_role, site=site_2, status=status_active +# ) + +# VLAN.objects.create(name="VLAN101", vid=101, status=status_active, site=site_1) + +# def test_data_loading(self): +# """Test the load() function.""" + +# job = IpFabricDataSource() +# job.job_result = JobResult.objects.create( +# name=job.class_path, obj_type=ContentType.objects.get_for_model(Job), user=None, job_id=uuid.uuid4() +# ) + +# nautobot = NautobotDiffSync( +# job=job, +# sync=None, +# safe_delete_mode=True, +# sync_ipfabric_tagged_only=True, +# ) +# nautobot.load() + +# self.assertEqual( +# set(["Site 1", "Site 2"]), +# {site.get_unique_id() for site in nautobot.get_all("location")}, +# ) +# self.assertEqual( +# set(["csr1", "csr2"]), +# {dev.get_unique_id() for dev in nautobot.get_all("device")}, +# ) + +# # Assert each site has a device tied to it. +# for device in nautobot.get_all("device"): +# self.assertTrue(hasattr(device, "location_name"), f"{device} is missing location_name") +# self.assertTrue(hasattr(device, "model"), f"{device} is missing model") +# self.assertTrue(hasattr(device, "name"), f"{device} is missing name") +# # These attributes don't exist on our Device DiffSyncModel yet but we may want them there in the future +# # self.assertTrue(hasattr(device, "platform")) +# # self.assertTrue(hasattr(device, "role")) +# # self.assertTrue(hasattr(device, "status")) +# self.assertTrue(hasattr(device, "serial_number"), f"{device} is missing serial_number") + +# # Assert each vlan has the necessary attributes +# for vlan in nautobot.get_all("vlan"): +# self.assertTrue(hasattr(vlan, "name")) +# self.assertTrue(hasattr(vlan, "vid")) +# self.assertTrue(hasattr(vlan, "status")) +# self.assertTrue(hasattr(vlan, "site")) diff --git a/nautobot_ssot/tests/ipfabric/test_nbutils.py b/nautobot_ssot/tests/ipfabric/test_nbutils.py new file mode 100644 index 000000000..3ac2ce7e5 --- /dev/null +++ b/nautobot_ssot/tests/ipfabric/test_nbutils.py @@ -0,0 +1,134 @@ +"""Test Nautobot Utilities.""" +from django.contrib.contenttypes.models import ContentType +from django.test import TestCase +from nautobot.dcim.models import DeviceType, Manufacturer, Location, LocationType +from nautobot.dcim.models.devices import Device +from nautobot.extras.models.statuses import Status +from nautobot.ipam.models import VLAN, IPAddress, Prefix, get_default_namespace +from nautobot.core.choices import ColorChoices +from nautobot.extras.models import Role + +from nautobot_ssot.integrations.ipfabric.utilities import ( # create_ip,; create_interface,; create_location, + get_or_create_device_role_object, + create_device_type_object, + create_manufacturer, + create_status, + create_vlan, +) + + +# pylint: disable=too-many-instance-attributes +class TestNautobotUtils(TestCase): + """Test Nautobot Utility.""" + + def setUp(self): + """Setup.""" + reg_loctype = LocationType.objects.update_or_create(name="Region")[0] + reg_loctype.content_types.set([ContentType.objects.get_for_model(VLAN)]) + self.location = Location.objects.create( + name="Test-Location", + status=Status.objects.get(name="Active"), + location_type=reg_loctype, + ) + + status_active = Status.objects.get(name="Active") + + self.manufacturer = Manufacturer.objects.create(name="Test-Manufacturer") + self.device_type = DeviceType.objects.create(model="Test-DeviceType", manufacturer=self.manufacturer) + self.content_type = ContentType.objects.get_for_model(Device) + self.device_role = Role.objects.create(name="Test-Role", color=ColorChoices.COLOR_RED) + self.device_role.content_types.set([self.content_type]) + self.device_role.cf["ipfabric_type"] = "Test-Role" + self.device_role.validated_save() + self.status = Status.objects.create( + name="Test-Status", + color=ColorChoices.COLOR_AMBER, + description="Test-Description", + ) + self.status.content_types.set([self.content_type]) + prefix = Prefix.objects.get_or_create( + prefix="192.168.0.0/16", namespace=get_default_namespace(), status=status_active + )[0] + self.ip_address = IPAddress.objects.create(address="192.168.0.1/32", status=status_active, parent=prefix) + + self.device = Device.objects.create( + name="Test-Device", + location=self.location, + device_type=self.device_type, + role=self.device_role, + status=status_active, + ) + + self.device.interfaces.create(name="Test-Interface", status=status_active) + self.vlan_content_type = ContentType.objects.get(app_label="ipam", model="vlan") + self.vlan_status = Status.objects.create( + name="Test-Vlan-Status", + color=ColorChoices.COLOR_AMBER, + description="Test-Description", + ) + self.vlan_status.content_types.set([self.vlan_content_type]) + + def test_create_vlan(self): + """Test `create_vlan` Utility.""" + vlan = create_vlan( + vlan_name="Test-Vlan", + vlan_id=100, + vlan_status="Test-Vlan-Status", + location_obj=self.location, + description="Test-Vlan", + ) + self.assertEqual(VLAN.objects.get(name="Test-Vlan").pk, vlan.pk) + + # def test_create_location(self): + # """Test `create_location` Utility.""" + # test_location = create_location(location_name="Test-Location") + # self.assertEqual(test_location.id, self.location.id) + + # def test_create_location_exception(self): + # """Test `create_location` Utility exception.""" + # location = create_location( + # location_name="Test-Location-100", + # location_id=123456, + # ) + # self.assertEqual(Location.objects.get(name="Test-Location-100").pk, location.pk) + + def test_create_device_type_object(self): + """Test `create_device_type_object` Utility.""" + test_device_type = create_device_type_object(device_type="Test-DeviceType", vendor_name="Test-Manufacturer") + self.assertEqual(test_device_type.id, self.device_type.id) + + def test_create_manufacturer(self): + """Test `create_manufacturer` Utility.""" + test_manufacturer = create_manufacturer(vendor_name="Test-Manufacturer") + self.assertEqual(test_manufacturer.id, self.manufacturer.id) + + def test_get_or_create_device_role(self): + """Test `get_or_create_device_role` Utility.""" + test_device_role = get_or_create_device_role_object("Test-Role", role_color=ColorChoices.COLOR_RED) + self.assertEqual(test_device_role.id, self.device_role.id) + + def test_create_status(self): + """Test `create_status` Utility.""" + test_status = create_status(status_name="Test-Status", status_color=ColorChoices.COLOR_AMBER) + self.assertEqual(test_status.id, self.status.id) + + def test_create_status_doesnt_exist(self): + """Test `create_status` Utility.""" + test_status = create_status(status_name="Test-Status-100", status_color=ColorChoices.COLOR_AMBER) + self.assertEqual(test_status.id, Status.objects.get(name="Test-Status-100").id) + + # def test_create_ip(self): + # """Test `create_ip` Utility.""" + # test_ip = create_ip("192.168.0.1", "255.255.255.255") + # self.assertEqual(test_ip.id, self.ip_address.id) + + # def test_create_ip_device_add(self): + # """Test `create_ip` adding to device Utility.""" + # test_ip = create_ip("192.168.0.1", "255.255.255.255", object_pk=self.device.id) + # self.assertEqual(test_ip.id, self.ip_address.id) + + # def test_create_interface(self): + # """Test `create_interface` Utility.""" + # interface_details = {"name": "Test-Interface"} + # test_interface = create_interface(self.device, interface_details) + # self.assertEqual(test_interface.id, self.device.interfaces.get(name="Test-Interface").id) diff --git a/poetry.lock b/poetry.lock index 1a6065fa2..51cd6983c 100644 --- a/poetry.lock +++ b/poetry.lock @@ -1,4 +1,4 @@ -# This file is automatically @generated by Poetry 1.5.1 and should not be changed by hand. +# This file is automatically @generated by Poetry 1.6.1 and should not be changed by hand. [[package]] name = "amqp" @@ -25,6 +25,27 @@ files = [ {file = "aniso8601-7.0.0.tar.gz", hash = "sha256:513d2b6637b7853806ae79ffaca6f3e8754bdd547048f5ccc1420aec4b714f1e"}, ] +[[package]] +name = "anyio" +version = "4.0.0" +description = "High level compatibility layer for multiple asynchronous event loop implementations" +optional = true +python-versions = ">=3.8" +files = [ + {file = "anyio-4.0.0-py3-none-any.whl", hash = "sha256:cfdb2b588b9fc25ede96d8db56ed50848b0b649dca3dd1df0b11f683bb9e0b5f"}, + {file = "anyio-4.0.0.tar.gz", hash = "sha256:f7ed51751b2c2add651e5747c891b47e26d2a21be5d32d9311dfe9692f3e5d7a"}, +] + +[package.dependencies] +exceptiongroup = {version = ">=1.0.2", markers = "python_version < \"3.11\""} +idna = ">=2.8" +sniffio = ">=1.1" + +[package.extras] +doc = ["Sphinx (>=7)", "packaging", "sphinx-autodoc-typehints (>=1.2.0)"] +test = ["anyio[trio]", "coverage[toml] (>=7)", "hypothesis (>=4.0)", "psutil (>=5.9)", "pytest (>=7.0)", "pytest-mock (>=3.6.1)", "trustme", "uvloop (>=0.17)"] +trio = ["trio (>=0.22)"] + [[package]] name = "appnope" version = "0.1.3" @@ -712,6 +733,24 @@ files = [ {file = "decorator-5.1.1.tar.gz", hash = "sha256:637996211036b6385ef91435e4fae22989472f9d571faba8927ba8253acbc330"}, ] +[[package]] +name = "deepdiff" +version = "6.5.0" +description = "Deep Difference and Search of any Python object/data. Recreate objects by adding adding deltas to each other." +optional = true +python-versions = ">=3.7" +files = [ + {file = "deepdiff-6.5.0-py3-none-any.whl", hash = "sha256:acdc1651a3e802415e0337b7e1192df5cd7c17b72fbab480466fdd799b9a72e7"}, + {file = "deepdiff-6.5.0.tar.gz", hash = "sha256:080b1359d6128f3f5f1738c6be3064f0ad9b0cc41994aa90a028065f6ad11f25"}, +] + +[package.dependencies] +ordered-set = ">=4.0.2,<4.2.0" + +[package.extras] +cli = ["click (==8.1.3)", "pyyaml (==6.0.1)"] +optimize = ["orjson"] + [[package]] name = "defusedxml" version = "0.7.1" @@ -1412,6 +1451,61 @@ files = [ [package.extras] protobuf = ["grpcio-tools (>=1.57.0)"] +[[package]] +name = "h11" +version = "0.14.0" +description = "A pure-Python, bring-your-own-I/O implementation of HTTP/1.1" +optional = true +python-versions = ">=3.7" +files = [ + {file = "h11-0.14.0-py3-none-any.whl", hash = "sha256:e3fe4ac4b851c468cc8363d500db52c2ead036020723024a109d37346efaa761"}, + {file = "h11-0.14.0.tar.gz", hash = "sha256:8f19fbbe99e72420ff35c00b27a34cb9937e902a8b810e2c88300c6f0a3b699d"}, +] + +[[package]] +name = "httpcore" +version = "0.16.3" +description = "A minimal low-level HTTP client." +optional = true +python-versions = ">=3.7" +files = [ + {file = "httpcore-0.16.3-py3-none-any.whl", hash = "sha256:da1fb708784a938aa084bde4feb8317056c55037247c787bd7e19eb2c2949dc0"}, + {file = "httpcore-0.16.3.tar.gz", hash = "sha256:c5d6f04e2fc530f39e0c077e6a30caa53f1451096120f1f38b954afd0b17c0cb"}, +] + +[package.dependencies] +anyio = ">=3.0,<5.0" +certifi = "*" +h11 = ">=0.13,<0.15" +sniffio = "==1.*" + +[package.extras] +http2 = ["h2 (>=3,<5)"] +socks = ["socksio (==1.*)"] + +[[package]] +name = "httpx" +version = "0.23.3" +description = "The next generation HTTP client." +optional = true +python-versions = ">=3.7" +files = [ + {file = "httpx-0.23.3-py3-none-any.whl", hash = "sha256:a211fcce9b1254ea24f0cd6af9869b3d29aba40154e947d2a07bb499b3e310d6"}, + {file = "httpx-0.23.3.tar.gz", hash = "sha256:9818458eb565bb54898ccb9b8b251a28785dd4a55afbc23d0eb410754fe7d0f9"}, +] + +[package.dependencies] +certifi = "*" +httpcore = ">=0.15.0,<0.17.0" +rfc3986 = {version = ">=1.3,<2", extras = ["idna2008"]} +sniffio = "*" + +[package.extras] +brotli = ["brotli", "brotlicffi"] +cli = ["click (==8.*)", "pygments (==2.*)", "rich (>=10,<13)"] +http2 = ["h2 (>=3,<5)"] +socks = ["socksio (==1.*)"] + [[package]] name = "idna" version = "3.4" @@ -1580,6 +1674,64 @@ files = [ {file = "invoke-2.2.0.tar.gz", hash = "sha256:ee6cbb101af1a859c7fe84f2a264c059020b0cb7fe3535f9424300ab568f6bd5"}, ] +[[package]] +name = "ipfabric" +version = "6.0.10" +description = "Python package for interacting with IP Fabric" +optional = true +python-versions = ">=3.7.1,<4.0.0" +files = [ + {file = "ipfabric-6.0.10-py3-none-any.whl", hash = "sha256:22a45ac36199eb69226dc9213d0d3f4df8d0e156ff7982c9a7dd6ff80409b598"}, + {file = "ipfabric-6.0.10.tar.gz", hash = "sha256:7dca65f77b14b5bda04ba76abbed0685b81d934e6b1f53a67c54bc8509ba9807"}, +] + +[package.dependencies] +deepdiff = ">=6.2.2,<7.0.0" +httpx = ">=0.23.2,<0.24.0" +ipfabric-httpx-auth = ">=6.0.0,<7.0.0" +macaddress = ">=2.0.2,<2.1.0" +pydantic = ">=1.8.2,<2.0.0" +python-dateutil = ">=2.8.2,<3.0.0" +python-dotenv = ">=0.21,<0.22" +pytz = ">=2022.4,<2023.0" + +[package.extras] +examples = ["openpyxl (>=3.0.9,<4.0.0)", "pandas (>=1.3.0,<2.0.0)", "python-json-logger (>=2.0.4,<3.0.0)", "pyyaml (>=6.0,<7.0)", "tabulate (>=0.8.9,<0.10.0)"] + +[[package]] +name = "ipfabric-diagrams" +version = "6.0.2" +description = "Python package for interacting with IP Fabric Diagrams" +optional = true +python-versions = ">=3.7.1,<4.0.0" +files = [ + {file = "ipfabric_diagrams-6.0.2-py3-none-any.whl", hash = "sha256:933e58250d7bd09ee5557667fe8bf4add1e9e3e1ec7dd137345f97bdd928b255"}, + {file = "ipfabric_diagrams-6.0.2.tar.gz", hash = "sha256:a76791e8547f48d70b59e5d653271f224b7482e5d77f995eca3be36886118900"}, +] + +[package.dependencies] +ipfabric = ">=6.0.7,<6.1.0" +pydantic = ">=1.8.2,<2.0.0" +typing-extensions = ">=4.1.1,<5.0.0" + +[package.extras] +examples = ["rich (>=12.5.1,<13.0.0)"] + +[[package]] +name = "ipfabric-httpx-auth" +version = "6.0.1" +description = "DEPRECATED: Authentication plugin for IP Fabric" +optional = true +python-versions = ">=3.7.1,<4.0.0" +files = [ + {file = "ipfabric_httpx_auth-6.0.1-py3-none-any.whl", hash = "sha256:7e47559d7a0f575f77b32ddefbeaff416973e36fe5728493c86a85f64ba918ba"}, + {file = "ipfabric_httpx_auth-6.0.1.tar.gz", hash = "sha256:32c713ed7a2326e9b11c581379e748b8893e24ea97e0388378a95e82b392bc0c"}, +] + +[package.dependencies] +httpx = ">=0.23.0,<0.24.0" +PyJWT = ">=2.4.0,<3.0.0" + [[package]] name = "ipython" version = "8.12.2" @@ -1789,6 +1941,17 @@ files = [ {file = "lazy_object_proxy-1.9.0-cp39-cp39-win_amd64.whl", hash = "sha256:db1c1722726f47e10e0b5fdbf15ac3b8adb58c091d12b3ab713965795036985f"}, ] +[[package]] +name = "macaddress" +version = "2.0.2" +description = "Like ``ipaddress``, but for hardware identifiers such as MAC addresses." +optional = true +python-versions = "*" +files = [ + {file = "macaddress-2.0.2-py3-none-any.whl", hash = "sha256:6f4a0430f9b5af6d98a582b8d527ba2cd3f0825fce5503a9ce5c73acb772c30f"}, + {file = "macaddress-2.0.2.tar.gz", hash = "sha256:1400ccdc28d747102d57ae61e5b78d8985872930810ceb8860cd49abd1e1fa37"}, +] + [[package]] name = "markdown" version = "3.3.7" @@ -2265,6 +2428,20 @@ rsa = ["cryptography (>=3.0.0)"] signals = ["blinker (>=1.4.0)"] signedtoken = ["cryptography (>=3.0.0)", "pyjwt (>=2.0.0,<3)"] +[[package]] +name = "ordered-set" +version = "4.1.0" +description = "An OrderedSet is a custom MutableSet that remembers its order, so that every" +optional = true +python-versions = ">=3.7" +files = [ + {file = "ordered-set-4.1.0.tar.gz", hash = "sha256:694a8e44c87657c59292ede72891eb91d34131f6531463aab3009191c77364a8"}, + {file = "ordered_set-4.1.0-py3-none-any.whl", hash = "sha256:046e1132c71fcf3330438a539928932caf51ddbc582496833e23de611de14562"}, +] + +[package.extras] +dev = ["black", "mypy", "pytest"] + [[package]] name = "packaging" version = "23.1" @@ -2972,13 +3149,13 @@ postgresql = ["psycopg2"] [[package]] name = "pytz" -version = "2023.3.post1" +version = "2022.7.1" description = "World timezone definitions, modern and historical" optional = false python-versions = "*" files = [ - {file = "pytz-2023.3.post1-py2.py3-none-any.whl", hash = "sha256:ce42d816b81b68506614c11e8937d3aa9e41007ceb50bfdcb0749b921bf646c7"}, - {file = "pytz-2023.3.post1.tar.gz", hash = "sha256:7b4fddbeb94a1eba4b557da24f19fdf9db575192544270a9101d8509f9f43d7b"}, + {file = "pytz-2022.7.1-py2.py3-none-any.whl", hash = "sha256:78f4f37d8198e0627c5f1143240bb0206b8691d8d7ac6d78fee88b78733f8c4a"}, + {file = "pytz-2022.7.1.tar.gz", hash = "sha256:01a0681c4b9684a28304615eba55d1ab31ae00bf68ec157ec3708a8182dbbcd0"}, ] [[package]] @@ -3285,6 +3462,23 @@ requests = ">=2.0.0" [package.extras] rsa = ["oauthlib[signedtoken] (>=3.0.0)"] +[[package]] +name = "rfc3986" +version = "1.5.0" +description = "Validating URI References per RFC 3986" +optional = true +python-versions = "*" +files = [ + {file = "rfc3986-1.5.0-py2.py3-none-any.whl", hash = "sha256:a86d6e1f5b1dc238b218b012df0aa79409667bb209e58da56d0b94704e712a97"}, + {file = "rfc3986-1.5.0.tar.gz", hash = "sha256:270aaf10d87d0d4e095063c65bf3ddbc6ee3d0b226328ce21e036f946e421835"}, +] + +[package.dependencies] +idna = {version = "*", optional = true, markers = "extra == \"idna2008\""} + +[package.extras] +idna2008 = ["idna"] + [[package]] name = "rich" version = "13.5.2" @@ -3457,6 +3651,17 @@ files = [ {file = "smmap-5.0.0.tar.gz", hash = "sha256:c840e62059cd3be204b0c9c9f74be2c09d5648eddd4580d9314c3ecde0b30936"}, ] +[[package]] +name = "sniffio" +version = "1.3.0" +description = "Sniff out which async library your code is running under" +optional = true +python-versions = ">=3.7" +files = [ + {file = "sniffio-1.3.0-py3-none-any.whl", hash = "sha256:eecefdce1e5bbfb7ad2eeaabf7c1eeb404d7757c379bd1f7e5cce9d8bf425384"}, + {file = "sniffio-1.3.0.tar.gz", hash = "sha256:e60305c5e5d314f5389259b7f22aaa33d8f7dee49763119234af3755c55b9101"}, +] + [[package]] name = "snowballstemmer" version = "2.2.0" @@ -3923,9 +4128,10 @@ testing = ["big-O", "jaraco.functools", "jaraco.itertools", "more-itertools", "p [extras] aci = ["PyYAML"] -all = ["Jinja2", "PyYAML", "cloudvision", "cvprac", "dnspython", "ijson", "oauthlib", "python-magic", "pytz", "requests", "requests-oauthlib", "six"] +all = ["Jinja2", "PyYAML", "cloudvision", "cvprac", "dnspython", "ijson", "ipfabric", "ipfabric-diagrams", "netutils", "oauthlib", "python-magic", "pytz", "requests", "requests-oauthlib", "six"] aristacv = ["cloudvision", "cvprac"] infoblox = ["dnspython"] +ipfabric = ["httpx", "ipfabric", "ipfabric-diagrams", "netutils"] nautobot-device-lifecycle-mgmt = [] pysnow = ["ijson", "oauthlib", "python-magic", "pytz", "requests", "requests-oauthlib", "six"] servicenow = ["Jinja2", "PyYAML", "ijson", "oauthlib", "python-magic", "pytz", "requests", "requests-oauthlib", "six"] @@ -3933,4 +4139,4 @@ servicenow = ["Jinja2", "PyYAML", "ijson", "oauthlib", "python-magic", "pytz", " [metadata] lock-version = "2.0" python-versions = "^3.8,<3.12" -content-hash = "15c124872e02b12f84f614cfd564d39883075b091c0268ed11b81f17ca1d14a0" +content-hash = "814f81470e03cc59f97b56eb8d33760113e3b2ffd011c6ebd135adec74b75b39" diff --git a/pyproject.toml b/pyproject.toml index 83781f150..947606bde 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -39,6 +39,9 @@ dnspython = { version = "^2.1.0", optional = true } packaging = ">=21.3, <24" prometheus-client = "~0.17.1" ijson = { version = ">=2.5.1", optional = true } +ipfabric = { version = "~6.0.9", optional = true } +ipfabric-diagrams = { version = "~6.0.2", optional = true } +netutils = { version = "^1.0.0", optional = true } oauthlib = { version = ">=3.1.0", optional = true } python-magic = { version = ">=0.4.15", optional = true } pytz = { version = ">=2019.3", optional = true } @@ -46,6 +49,7 @@ requests = { version = ">=2.21.0", optional = true } requests-oauthlib = { version = ">=1.3.0", optional = true } six = { version = ">=1.13.0", optional = true } drf-spectacular = "0.26.3" +httpx = { version = ">=0.23.3", optional = true } [tool.poetry.dev-dependencies] bandit = "*" @@ -79,6 +83,8 @@ mkdocstrings = "0.22.0" mkdocstrings-python = "1.1.2" requests-mock = "^1.10.0" parameterized = "^0.8.1" +# Enable chatops after dropping Python 3.7 support +# nautobot-chatops = { version = "^2.0.2", extras = ["ipfabric"] } [tool.poetry.plugins."nautobot_ssot.data_sources"] "example" = "nautobot_ssot.sync.example:ExampleSyncWorker" @@ -86,6 +92,10 @@ parameterized = "^0.8.1" [tool.poetry.plugins."nautobot_ssot.data_targets"] "example" = "nautobot_ssot.sync.example:ExampleSyncWorker" +# Enable chatops after dropping Python 3.7 support +# [tool.poetry.plugins."nautobot.workers"] +# "ipfabric" = "nautobot_ssot.integrations.ipfabric.workers:ipfabric" + [tool.poetry.extras] aci = [ "PyYAML", @@ -97,7 +107,10 @@ all = [ "cvprac", "dnspython", "ijson", + "ipfabric", + "ipfabric-diagrams", "nautobot-device-lifecycle-mgmt", + "netutils", "oauthlib", "python-magic", "pytz", @@ -112,6 +125,12 @@ aristacv = [ infoblox = [ "dnspython", ] +ipfabric = [ + "httpx", + "ipfabric", + "ipfabric-diagrams", + "netutils", +] # pysnow = "^0.7.17" # PySNow is currently pinned to an older version of pytz as a dependency, which blocks compatibility with newer # versions of Nautobot. See https://github.com/rbw/pysnow/pull/186