Add Arista/EOS support

.gitignore

@@ -14,4 +14,5 @@ machines/**
 
 .vscode
 .env
-.code
+.venv/
+.code

cosmo/clients/netbox_v4.py

@@ -103,25 +103,27 @@ def _merge_into(self, data: dict, query_data):
         loopbacks: dict[str, dict] = dict()
 
         for interface in query_data["interface_list"]:
+            device_name = interface["device"]["name"]
+
             child_interface = next(
                 filter(lambda i: i["vrf"] is None, interface["child_interfaces"]), None
             )
-            if not child_interface:
-                continue
-            device_name = interface["device"]["name"]
+            candidate_ip_addresses = []
+            if child_interface:
+                candidate_ip_addresses = child_interface.get("ip_addresses", [])
+            elif interface.get("vrf") is None:
+                candidate_ip_addresses = interface.get("ip_addresses", [])
 
             l_ipv4 = next(
-                filter(
-                    lambda l: l["family"]["value"] == 4, child_interface["ip_addresses"]
-                ),
+                filter(lambda l: l["family"]["value"] == 4, candidate_ip_addresses),
                 None,
             )
             l_ipv6 = next(
-                filter(
-                    lambda l: l["family"]["value"] == 6, child_interface["ip_addresses"]
-                ),
+                filter(lambda l: l["family"]["value"] == 6, candidate_ip_addresses),
                 None,
             )
+            if not l_ipv4 and not l_ipv6:
+                continue
             loopbacks[device_name] = {
                 "ipv4": l_ipv4["address"] if l_ipv4 else None,
                 "ipv6": l_ipv6["address"] if l_ipv6 else None,

cosmo/clients/queries/loopback.graphql

@@ -1,9 +1,32 @@
 query{
     interface_list(filters: {
-        name: {starts_with: "lo"}
+        name: {i_starts_with: "lo"}
     }) {
         __typename
         name,
+        vrf {
+            __typename
+            id
+            name
+            description
+            rd
+            export_targets {
+                __typename
+                name
+            }
+            import_targets {
+                __typename
+                name
+            }
+        },
+        ip_addresses {
+            __typename
+            address,
+            family {
+                __typename
+                value,
+            }
+        }
         child_interfaces {
             __typename
             name,

cosmo/manufacturers.py

@@ -103,6 +103,10 @@ def getRibTableNameFor(self, v: VRFType, af: int) -> str:
     def hasMTUInheritance():
         pass
 
+    @staticmethod
+    def supportsDirectInterfaceIP():
+        return False
+
 
 class AbstractJuniperRtBrickManufacturerCommon(AbstractManufacturer, ABC):
     VRF_KEY: Final[str] = "routing_instances"
@@ -186,6 +190,27 @@ def _spitDefaultVRFPathWith(cls, d: dict) -> dict:
         return {cls.VRF_KEY: {cls.DEFAULT_VRF_KEY: {**d}}}
 
 
+class AristaManufacturer(RtBrickManufacturer):
+    _platform_re = re.compile(r"^(arista-)?eos(64)?(-[a-zA-Z0-9-]*)?$")
+
+    @staticmethod
+    def myManufacturerSlugs():
+        return ["arista", "arista-networks"]
+
+    @staticmethod
+    def getManagementVRFName():
+        return "MGMT"
+
+    def isManagementInterface(self, o: InterfaceType):
+        return len(o["ip_addresses"]) >= 1 and o.getName().lower().startswith(
+            "management"
+        )
+
+    @staticmethod
+    def supportsDirectInterfaceIP():
+        return True
+
+
 class CumulusNetworksManufacturer(AbstractManufacturer):
     _platform_re = re.compile(r"^cumulus-linux[a-zA-Z0-9-]*")
 
@@ -228,6 +253,7 @@ def getRibTableNameFor(self, v: VRFType, af: int) -> str:
 class ManufacturerFactoryFromDevice:
     _all_manufacturers = (
         CumulusNetworksManufacturer,
+        AristaManufacturer,
         RtBrickManufacturer,
         JuniperManufacturer,
     )
@@ -237,8 +263,34 @@ def __init__(self, device: DeviceType, cosmo_config: "CosmoConfig"):
         self._device = device
         self._cosmo_config = cosmo_config
 
+    @staticmethod
+    def _manufacturerSlug(o: DeviceTypeType | PlatformType):
+        if isinstance(o, dict):
+            manufacturer = o.get("manufacturer")
+            if isinstance(manufacturer, dict):
+                return manufacturer.get("slug")
+            return None
+        manufacturer = o.getManufacturer()
+        if manufacturer:
+            return manufacturer.get("slug")
+        return None
+
+    @staticmethod
+    def _slug(o: DeviceTypeType | PlatformType):
+        if isinstance(o, dict):
+            return o.get("slug")
+        return o.get("slug")
+
     def get(self) -> AbstractManufacturer | NoReturn:
         for c in self._all_manufacturers:
             if c.isCompatibleWith(self._device):
                 return c(self._cosmo_config)
-        raise Exception(f"Cannot find suitable manufacturer for device {self._device}")
+        device_type = self._device.getDeviceType()
+        platform = self._device.getPlatform()
+        raise Exception(
+            f"Cannot find suitable manufacturer for device {self._device}. "
+            f"device_type_slug={self._slug(device_type)}, "
+            f"device_type_manufacturer_slug={self._manufacturerSlug(device_type)}, "
+            f"platform_slug={self._slug(platform)}, "
+            f"platform_manufacturer_slug={self._manufacturerSlug(platform)}"
+        )

cosmo/netbox_types.py

@@ -281,6 +281,17 @@ def getPlatform(self):
     def getInterfaces(self) -> list["InterfaceType"]:
         return self.get("interfaces", [])
 
+    def deriveRouterIdFromLoopbackInterface(self) -> str | None:
+        for interface in self.getInterfaces():
+            if interface.getVRF() is not None:
+                continue
+            if not interface.isLoopbackOrParentIsLoopback():
+                continue
+            for address in interface.getIPAddresses():
+                if address.getIPInterfaceObject().version == 4:
+                    return str(address.getIPInterfaceObject().ip)
+        return None
+
     def getISISIdentifier(self) -> str | None | Never:
         sys_id: Any | None = self.getCustomFields().get("isis_system_id")
         if sys_id and not re.match(r"\d{4}.\d{4}.\d{4}", str(sys_id)):
@@ -693,7 +704,7 @@ def isLoopbackOrParentIsLoopback(self) -> bool:
                     on=self,
                 )
             return parent_interface.isLoopbackOrParentIsLoopback()
-        elif self.getName().startswith("lo"):
+        elif self.getName().lower().startswith("lo"):
             if self.getAssociatedType() != "loopback" and features.featureIsEnabled(
                 "netbox-loopback-interface-type"
             ):

cosmo/routerbgpcpevisitor.py

@@ -158,19 +158,22 @@ def processBgpCpeTag(self, o: TagType):
         manufacturer = ManufacturerFactoryFromDevice(
             o.getParent(DeviceType), self._cosmo_config
         ).get()
-        if not linked_interface.hasParentInterface():
+        if linked_interface.hasParentInterface():
+            parent_interface = next(
+                filter(
+                    lambda interface: interface == linked_interface["parent"],
+                    o.getParent(DeviceType).getInterfaces(),
+                )
+            )
+        elif manufacturer.supportsDirectInterfaceIP():
+            parent_interface = linked_interface
+        else:
             warn(
                 f"does not have a parent interface configured, skipping...",
                 linked_interface,
             )
             return
 
-        parent_interface = next(
-            filter(
-                lambda interface: interface == linked_interface["parent"],
-                o.getParent(DeviceType).getInterfaces(),
-            )
-        )
         cpe = head(parent_interface.getConnectedEndpoints())
         if not cpe:
             warn(

cosmo/routervisitor.py

@@ -146,7 +146,11 @@ def _(self, o: IPAddressType):
         ).get()
         optional_attrs = {}
         parent_interface = o.getParent(InterfaceType)
-        if not parent_interface.isSubInterface():
+        is_management_interface = manufacturer.isManagementInterface(parent_interface)
+        if (
+            not parent_interface.isSubInterface()
+            and not manufacturer.supportsDirectInterfaceIP()
+        ):
             raise InterfaceSerializationError(
                 f"You seem to have configured an IP directly on interface {parent_interface.getName()}. "
                 f"This is forbidden. Please make a virtual interface, assign the IP(s) on it and retry!"
@@ -155,16 +159,13 @@ def _(self, o: IPAddressType):
             o.isGlobal()
             or self.allow_private_ips
             or parent_interface.getVRF()
-            or manufacturer.isManagementInterface(parent_interface)
+            or is_management_interface
         ):
             raise InterfaceSerializationError(
                 f"Private IP {o.getIPAddress()} used on interface {parent_interface.getName()} "
                 f"in default VRF for device {o.getParent(DeviceType).getName()}. Did you forget to configure a VRF?"
             )
-        if (
-            manufacturer.isManagementInterface(parent_interface)
-            and parent_interface.isSubInterface()
-        ):
+        if is_management_interface:
             optional_attrs = self.processMgmtInterfaceIPAddress(o)
         if (
             parent_interface.isLoopbackOrParentIsLoopback()
@@ -202,9 +203,14 @@ def _(self, o: IPAddressType):
         # are not mgmt or loopback interfaces
         if (
             parent_interface.getVRF() == None
-            and not manufacturer.isManagementInterface(parent_interface)
+            and not is_management_interface
             and not parent_interface.isLoopbackOrParentIsLoopback()
-            and not any([t.getTagName() == "disable_sampling" for t in parent_interface.getTags()])
+            and not any(
+                [
+                    t.getTagName() == "disable_sampling"
+                    for t in parent_interface.getTags()
+                ]
+            )
         ):
             sampling = {"sampling": True}
         return {
@@ -372,11 +378,31 @@ def _(self, o: VRFType):
             parent_device, self._cosmo_config
         ).get()
 
-        if not parent_interface.isSubInterface():
+        if (
+            not parent_interface.isSubInterface()
+            and not manufacturer.supportsDirectInterfaceIP()
+        ):
             return  # guard: do not process root interface
 
-        loopback = self.loopbacks.getByDevice(parent_device.getName())
-        router_id = loopback.deriveRouterId()
+        if (
+            manufacturer.supportsDirectInterfaceIP()
+            and not parent_interface.isSubInterface()
+            and not o.isMgmtVRF()
+            and not manufacturer.isGlobalVRF(o)
+            and not parent_interface.getIPAddresses()
+            and not self.interfaceHasTag(parent_interface, "unnumbered")
+            and not self.interfaceHasTag(parent_interface, "unnumbered0")
+        ):
+            raise InterfaceSerializationError(
+                f"Interface {parent_interface.getName()} is assigned to VRF {o.getName()} "
+                "but has no IP address or unnumbered tag.",
+                on=parent_interface,
+            )
+
+        router_id = parent_device.deriveRouterIdFromLoopbackInterface()
+        if not router_id:
+            loopback = self.loopbacks.getByDevice(parent_device.getName())
+            router_id = loopback.deriveRouterId()
         if o.getRouteDistinguisher():
             rd = router_id + ":" + o.getRouteDistinguisher()
         elif not o.isMgmtVRF():
@@ -406,6 +432,10 @@ def _(self, o: VRFType):
             },
         )
 
+    @staticmethod
+    def interfaceHasTag(o: InterfaceType, name: str) -> bool:
+        return any(tag.getTagName() == name for tag in o.getTags())
+
     @staticmethod
     def processStaticRouteCommon(o: CosmoStaticRouteType, m: AbstractManufacturer):
         next_hop = None
@@ -645,15 +675,18 @@ def processCoreTag(self, o: TagType):
         ).get()
 
         interface = o.getParent(InterfaceType)
-        parent_interface = head(
-            list(
-                filter(  # as in, netbox parent
-                    lambda i: i.getName()
-                    == interface.getSubInterfaceParentInterfaceName(),
-                    interface.getParent(DeviceType).getInterfaces(),
+        if manufacturer.supportsDirectInterfaceIP() and not interface.isSubInterface():
+            parent_interface = interface
+        else:
+            parent_interface = head(
+                list(
+                    filter(  # as in, netbox parent
+                        lambda i: i.getName()
+                        == interface.getSubInterfaceParentInterfaceName(),
+                        interface.getParent(DeviceType).getInterfaces(),
+                    )
                 )
             )
-        )
 
         # Note:
         # The following code was developed by the pseudo code:
@@ -759,16 +792,22 @@ def processBgpUnnumberedTag(self, o: TagType, prefer_unit0=False):
         # Therefore, there is a unnumbered0 Tag for backwards compat and a unnumbered tag.
         # This method handles both of them, for unnumbered0 prefer_unit0 is true.
 
+        manufacturer = ManufacturerFactoryFromDevice(
+            o.getParent(DeviceType), self._cosmo_config
+        ).get()
+
         def loopback_filter_function(i, parent_interface: InterfaceType):
 
-            if prefer_unit0:
+            if manufacturer.supportsDirectInterfaceIP() and not i.isSubInterface():
+                is_correct_unit = True
+            elif prefer_unit0:
                 is_correct_unit = i.getUnitNumber() == 0
             else:
                 is_correct_unit = i.getUnitNumber() != 0
 
             return (
-                i.getName().startswith("lo")
-                and i.isSubInterface()
+                i.getName().lower().startswith("lo")
+                and (i.isSubInterface() or manufacturer.supportsDirectInterfaceIP())
                 and i.getVRF() == parent_interface.getVRF()
                 and is_correct_unit
             )
@@ -782,6 +821,11 @@ def loopback_filter_function(i, parent_interface: InterfaceType):
                 )
             )
         )
+        if not loopback_interface:
+            raise InterfaceSerializationError(
+                f"Cannot find a suitable loopback interface for unnumbered interface {parent_interface.getName()}.",
+                on=parent_interface,
+            )
         opt_unnumbered_interface = {
             "unnumbered_interface": loopback_interface.getName()
         }