feat: Nautobot interface sync

.github/workflows/build-container-images.yaml

@@ -23,6 +23,7 @@ env:
   VERSION_PYTHON_IRONIC: 0.0.1
   VERSION_ARGO_UTILS: 0.0.1
   VERSION_OBM_UTILS: 0.0.1
+  VERSION_PYTHON_NAUTOBOT_INT_SYNC: 0.0.1
 
 jobs:
   build-ghcr-registry:
@@ -97,3 +98,14 @@ jobs:
           tags: ghcr.io/rackerlabs/understack/argo-obm-utils-python3.11.8:latest,ghcr.io/rackerlabs/understack/argo-obm-utils-python3.11.8:${{ env.VERSION_OBM_UTILS }}
           labels: |
               org.opencontainers.image.version=${{ env.VERSION_OBM_UTILS }}
+
+      - name: Build and deploy Python 3.11 with Nautobot Int Sync
+        uses: docker/build-push-action@v5
+        with:
+          context: argo-workflows/nautobot-interface-sync
+          file: argo-workflows/nautobot-interface-sync/containers/Dockerfile.nautobot_int_sync
+          # push for all main branch commits
+          push: ${{ github.event_name != 'pull_request' }}
+          tags: ghcr.io/rackerlabs/understack/nautobot-interfaces-sync:latest,ghcr.io/rackerlabs/understack/nautobot-interfaces-sync:${{ env.VERSION_PYTHON_NAUTOBOT_INT_SYNC }}
+          labels: |
+              org.opencontainers.image.version=${{ env.VERSION_PYTHON_NAUTOBOT_INT_SYNC }}

argo-workflows/nautobot-interface-sync/code/helpers.py

@@ -0,0 +1,63 @@
+import argparse
+import logging
+import os
+import sushy
+import sys
+
+logger = logging.getLogger(__name__)
+
+
+def setup_logger(name):
+    logger = logging.getLogger(name)
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter(
+        "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+    )
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(logging.DEBUG)
+    return logger
+
+
+def arg_parser(name):
+    parser = argparse.ArgumentParser(
+        prog=os.path.basename(name), description="Nautobot Interface sync"
+    )
+    parser.add_argument("--hostname", required=True,
+                        help="Nautobot device name")
+    parser.add_argument("--oob_username", required=False, help="OOB username")
+    parser.add_argument("--oob_password", required=False, help="OOB password")
+    parser.add_argument("--nautobot_url", required=False)
+    parser.add_argument("--nautobot_token", required=False)
+    return parser
+
+
+def exit_with_error(error):
+    logger.error(error)
+    sys.exit(1)
+
+
+def credential(subpath, item):
+    try:
+        return open(f"/etc/{subpath}/{item}", "r").read().strip()
+    except FileNotFoundError:
+        exit_with_error(f"{subpath} {item} not found in mounted files")
+
+
+def oob_sushy_session(oob_ip, oob_username, oob_password):
+    try:
+        return sushy.Sushy(
+            f"https://{oob_ip}",
+            username=oob_username,
+            password=oob_password,
+            verify=False,
+        )
+    except sushy.exceptions.ConnectionError as e:
+        exit_with_error(e)
+
+
+def is_off_board(interface):
+    return (
+        "Embedded" not in interface.location
+        and "Integrated" not in interface.name
+    )

argo-workflows/nautobot-interface-sync/code/models.py

@@ -0,0 +1,146 @@
+from __future__ import annotations
+from dataclasses import dataclass
+from sushy import Sushy
+from sushy.resources.system.network.adapter import NetworkAdapter
+from sushy.resources.system.network.port import NetworkPort
+from sushy.resources.chassis.chassis import Chassis as SushyChassis
+
+
+class ManufacturerNotSupported(Exception):
+    pass
+
+
+@dataclass
+class NIC:
+    name: str
+    location: str
+    interfaces: list[Interface]
+    model: str
+
+    @classmethod
+    def from_redfish(cls, data: NetworkAdapter) -> NIC:
+        location = cls.nic_location(data)
+        nic = cls(data.identity, location, [], data.model)
+        nic.interfaces = [
+            Interface.from_redfish(i, nic)
+            for i in cls.nic_ports(data)
+        ]
+        return nic
+
+    @classmethod
+    def from_hp_json(cls, data: dict) -> NIC:
+        nic = cls(data.get("name"), data.get("location"), [], data.get("name"))
+        ports = data.get("network_ports") or data.get("unknown_ports")
+        nic.interfaces = [Interface.from_hp_json(i, nic, ports) for i in ports]
+        return nic
+
+    @classmethod
+    def nic_location(cls, nic: NetworkAdapter) -> str:
+        try:
+            return nic.json["Controllers"][0]["Location"]["PartLocation"][
+                "ServiceLabel"
+            ]
+        except KeyError:
+            return nic.identity
+
+    @classmethod
+    def nic_ports(cls, nic: NetworkAdapter) -> list[NetworkPort]:
+        return nic.network_ports.get_members()
+
+
+@dataclass
+class Interface:
+    name: str
+    mac_addr: str
+    location: str
+    current_speed_mbps: int
+    nic_model: str
+
+    @classmethod
+    def from_redfish(cls, data: NetworkPort, nic: NIC) -> Interface:
+        if data.root.json["Vendor"] == "HPE":
+            name = f"{nic.name}_{data.physical_port_number}"
+        else:
+            name = data.identity
+        return cls(
+            name,
+            data.associated_network_addresses[0],
+            nic.location,
+            data.current_link_speed_mbps,
+            nic.model,
+        )
+
+    @classmethod
+    def from_hp_json(cls, data: dict, nic: NIC, ports: list) -> Interface:
+        p_num = data.get("port_num") or (ports.index(data) + 1)
+        interface_name = f"NIC.{nic.location.replace(' ', '.')}_{p_num}"
+        return cls(
+            interface_name,
+            data.get("mac_addr"),
+            nic.location,
+            data.get("speed", 0),
+            nic.model,
+        )
+
+
+@dataclass
+class Chassis:
+    name: str
+    nics: list[NIC]
+    network_interfaces: list[Interface]
+
+    @classmethod
+    def check_manufacturer(cls, manufacturer: str) -> None:
+        supported_manufacturers = ["HPE", "Dell Inc."]
+        if manufacturer not in supported_manufacturers:
+            raise ManufacturerNotSupported(
+                f"Manufacturer {manufacturer} not supported. "
+                f"Supported manufacturers: {', '.join(supported_manufacturers)}"
+            )
+
+    @classmethod
+    def obm_is_ilo4(cls, chassis_data: SushyChassis) -> bool:
+        return (
+            chassis_data.redfish_version == "1.0.0"
+            and chassis_data.manufacturer == "HPE"
+        )
+
+    @classmethod
+    def from_redfish(cls, oob_obj: Sushy) -> Chassis:
+        chassis_data = oob_obj.get_chassis(
+            oob_obj.get_chassis_collection().members_identities[0]
+        )
+
+        cls.check_manufacturer(chassis_data.manufacturer)
+
+        if cls.obm_is_ilo4(chassis_data):
+            return cls.from_hp_json(oob_obj, chassis_data.name)
+
+        chassis = cls(chassis_data.name, [], [])
+        chassis.nics = [
+            NIC.from_redfish(i)
+            for i in chassis_data.network_adapters.get_members()
+        ]
+        chassis.network_interfaces = cls.interfaces_from_nics(chassis.nics)
+        return chassis
+
+    @classmethod
+    def from_hp_json(cls, oob_obj: Sushy, chassis_name: str) -> Chassis:
+        data = cls.chassis_hp_json_data(oob_obj)
+        nics = [NIC.from_hp_json(i) for i in data]
+        network_interfaces = cls.interfaces_from_nics(nics)
+        return cls(chassis_name, nics, network_interfaces)
+
+    @classmethod
+    def interfaces_from_nics(cls, nics: list[NIC]) -> list[Interface]:
+        return [interface for nic in nics for interface in nic.interfaces]
+
+    @classmethod
+    def chassis_hp_json_data(cls, oob_obj: Sushy) -> dict:
+        oob_obj._conn.set_http_basic_auth(
+            username=oob_obj._auth._username, password=oob_obj._auth._password
+        )
+        resp = oob_obj._conn.get(path="/json/comm_controller_info")
+        resp.raise_for_status()
+        data = resp.json()["comm_controllers"]
+        return data

argo-workflows/nautobot-interface-sync/code/nautobot.py

@@ -0,0 +1,126 @@
+import logging
+import pynautobot
+import requests
+import sys
+from typing import Protocol
+from pynautobot.core.api import Api as NautobotApi
+from pynautobot.models.dcim import Devices as NautobotDevice
+from pynautobot.models.dcim import Interfaces as NautobotInterface
+
+
+class Interface(Protocol):
+    name: str
+    mac_addr: str
+    location: str
+
+
+class Nautobot:
+    def __init__(self, url, token, logger=None, session=None):
+        self.url = url
+        self.token = token
+        self.logger = logger or logging.getLogger(__name__)
+        self.session = session or self.api_session(self.url, self.token)
+
+    def exit_with_error(self, error):
+        self.logger.error(error)
+        sys.exit(1)
+
+    def api_session(self, url: str, token: str) -> NautobotApi:
+        try:
+            return pynautobot.api(url, token=token)
+        except requests.exceptions.ConnectionError as e:
+            self.exit_with_error(e)
+        except pynautobot.core.query.RequestError as e:
+            self.exit_with_error(e)
+
+    def device(self, device_name: str) -> NautobotDevice:
+        device = self.session.dcim.devices.get(name=device_name)
+        if not device:
+            self.exit_with_error(f"Device {device_name} not found in Nautobot")
+        return device
+
+    def device_oob_interface(
+        self,
+        device: NautobotDevice,
+    ) -> NautobotInterface:
+
+        oob_intf = self.session.dcim.interfaces.get(
+            device_id=device.id, name=["iDRAC", "iLO"]
+        )
+        if not oob_intf:
+            self.exit_with_error(
+                f"No OOB interfaces found for {device.name} in Nautobot"
+            )
+        return oob_intf
+
+    def ip_from_interface(self, interface: NautobotInterface) -> str:
+        ips = interface.ip_addresses
+        if not ips:
+            self.exit_with_error(
+                f"No IP addresses found for interface: {interface.name}"
+            )
+        return ips[0].host
+
+    def device_oob_ip(self, device_name: str) -> str:
+        device = self.device(device_name)
+        oob_intf = self.device_oob_interface(device)
+        oob_ip = self.ip_from_interface(oob_intf)
+        return oob_ip
+
+    def construct_interfaces_payload(
+        self,
+        interfaces: list[Interface],
+        device_id: str,
+        device_name: str,
+    ) -> list[dict]:
+
+        payload = []
+        for interface in interfaces:
+            nautobot_intf = self.session.dcim.interfaces.get(
+                device_id=device_id, name=interface.name
+            )
+            if nautobot_intf is None:
+                self.logger.info(
+                    f"{interface.name} was NOT found for "
+                    f"{device_name}, creating..."
+                )
+                payload.append(
+                    self.interface_payload_data(device_id, interface)
+                )
+            else:
+                self.logger.info(
+                    f"{nautobot_intf.name} found in Nautobot for "
+                    f"{device_name}, no action will be taken."
+                )
+        return payload
+
+    def interface_payload_data(
+        self, device_id: str, interface: Interface
+    ) -> dict:
+
+        return {
+            "device": device_id,
+            "name": interface.name,
+            "mac_address": interface.mac_addr,
+            "type": "other",
+            "status": "Active",
+            "description": f"Location: {interface.location}",
+        }
+
+    def bulk_create_interfaces(
+        self, device_name: str, interfaces: list[Interface]
+    ) -> list[NautobotInterface] | None:
+        device = self.device(device_name)
+        payload = self.construct_interfaces_payload(
+            interfaces, device.id, device.name
+        )
+        if payload:
+            try:
+                req = self.session.dcim.interfaces.create(payload)
+            except pynautobot.core.query.RequestError as e:
+                self.exit_with_error(e)
+
+            for interface in req:
+                self.logger.info(f"{interface.name} successfully created")
+
+            return req