diff --git a/.gitignore b/.gitignore index 107b4ca3..d2e8c95d 100644 --- a/.gitignore +++ b/.gitignore @@ -14,4 +14,5 @@ machines/** .vscode .env -.code \ No newline at end of file +.venv/ +.code diff --git a/cosmo/clients/netbox_v4.py b/cosmo/clients/netbox_v4.py index faac1018..b18f69eb 100644 --- a/cosmo/clients/netbox_v4.py +++ b/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, diff --git a/cosmo/clients/queries/loopback.graphql b/cosmo/clients/queries/loopback.graphql index b65f4f8f..b9e32c98 100644 --- a/cosmo/clients/queries/loopback.graphql +++ b/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, diff --git a/cosmo/manufacturers.py b/cosmo/manufacturers.py index 003023b9..9623aaf4 100644 --- a/cosmo/manufacturers.py +++ b/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)}" + ) diff --git a/cosmo/netbox_types.py b/cosmo/netbox_types.py index 2833eb1f..ae75a095 100644 --- a/cosmo/netbox_types.py +++ b/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" ): diff --git a/cosmo/routerbgpcpevisitor.py b/cosmo/routerbgpcpevisitor.py index cf45a829..4293efcf 100644 --- a/cosmo/routerbgpcpevisitor.py +++ b/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( diff --git a/cosmo/routervisitor.py b/cosmo/routervisitor.py index 220a8188..5bb4ae03 100644 --- a/cosmo/routervisitor.py +++ b/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() } diff --git a/cosmo/tests/test_serializer.py b/cosmo/tests/test_serializer.py index db136e0c..4bab43fc 100644 --- a/cosmo/tests/test_serializer.py +++ b/cosmo/tests/test_serializer.py @@ -7,7 +7,7 @@ from cosmo.common import DeviceSerializationError from cosmo.config.cosmo_config import CosmoConfig from cosmo.features import with_feature, features, without_feature -from cosmo.manufacturers import ManufacturerFactoryFromDevice +from cosmo.manufacturers import AristaManufacturer, ManufacturerFactoryFromDevice from cosmo.netbox_types import DeviceType, VRFType from coverage.html import os @@ -181,6 +181,258 @@ def test_router_platforms(mock_cosmo_config_fixture, mock_global_vrf, mock_l3vpn s.serialize() +def test_arista_manufacturer(mock_cosmo_config_fixture): + test_data = _yaml_load("./test_case_l3vpn.yml") + device = test_data["device_list"][0] + device["platform"]["manufacturer"]["slug"] = "arista" + device["platform"]["slug"] = "arista-eos64-4-33-8m" + management_interface = copy.deepcopy(device["interfaces"][0]) + management_interface.update( + { + "id": "management1", + "name": "Management1", + "ip_addresses": [ + { + "__typename": "IPAddressType", + "address": "192.0.2.2/24", + } + ], + } + ) + ethernet_interface = copy.deepcopy(device["interfaces"][0]) + ethernet_interface.update( + { + "id": "ethernet1", + "name": "Ethernet1", + "mtu": 9216, + "ip_addresses": [ + { + "__typename": "IPAddressType", + "address": "198.51.100.0/31", + } + ], + "tags": [ + { + "__typename": "TagType", + "name": "core", + "slug": "core", + } + ], + } + ) + unnumbered_interface = copy.deepcopy(device["interfaces"][0]) + unnumbered_interface.update( + { + "id": "ethernet2", + "name": "Ethernet2", + "ip_addresses": [], + "tags": [ + { + "__typename": "TagType", + "name": "unnumbered", + "slug": "unnumbered", + } + ], + } + ) + loopback_interface = copy.deepcopy(device["interfaces"][0]) + loopback_interface.update( + { + "id": "loopback0", + "name": "Loopback0", + "ip_addresses": [], + "tags": [], + "type": "LOOPBACK", + } + ) + device["interfaces"] = [ + interface + for interface in device["interfaces"] + if not interface["name"].startswith("lo-") + ] + device["interfaces"].extend( + [ + management_interface, + ethernet_interface, + unnumbered_interface, + loopback_interface, + ] + ) + + manufacturer = ManufacturerFactoryFromDevice( + DeviceType(device), mock_cosmo_config_fixture + ).get() + + assert isinstance(manufacturer, AristaManufacturer) + + serialized = ( + RouterSerializer( + device=device, + l2vpn_list=test_data["l2vpn_list"], + loopbacks=test_data.get("loopbacks", {}), + cosmo_config=mock_cosmo_config_fixture, + ) + .allowPrivateIPs() + .serialize() + ) + assert serialized["platform"] == "arista-eos64-4-33-8m" + assert serialized["routing_instances"]["MGMT"]["description"] == ( + "MGMT-ROUTING-INSTANCE" + ) + assert serialized["interfaces"]["Management1"]["families"]["inet"]["address"] == { + "192.0.2.2/24": {} + } + assert ( + serialized["routing_instances"]["MGMT"]["routing_options"]["rib"][ + "MGMT.inet.0" + ]["static"]["0.0.0.0/0"]["next_hop"] + == "192.0.2.1" + ) + assert serialized["interfaces"]["Ethernet1"]["families"]["inet"]["address"] == { + "198.51.100.0/31": {} + } + assert serialized["interfaces"]["Ethernet1"]["families"]["iso"] == {} + assert serialized["interfaces"]["Ethernet1"]["families"]["mpls"] == {} + assert serialized["interfaces"]["Ethernet1"]["mtu"] == 9216 + assert serialized["interfaces"]["Ethernet2"]["unnumbered"] is True + assert serialized["interfaces"]["Ethernet2"]["unnumbered_interface"] == "Loopback0" + + +def _arista_l3vpn_test_data(): + test_data = _yaml_load("./test_case_l3vpn.yml") + device = test_data["device_list"][0] + device["platform"]["manufacturer"]["slug"] = "arista" + device["platform"]["slug"] = "arista-eos64-4-33-8m" + return test_data, device + + +def _serialize_router_test_data(test_data, device, cosmo_config, loopbacks=None): + return ( + RouterSerializer( + device=device, + l2vpn_list=test_data["l2vpn_list"], + loopbacks=( + test_data.get("loopbacks", {}) if loopbacks is None else loopbacks + ), + cosmo_config=cosmo_config, + ) + .allowPrivateIPs() + .serialize() + ) + + +def test_arista_direct_lag_l3vpn(mock_cosmo_config_fixture): + test_data, device = _arista_l3vpn_test_data() + physical = device["interfaces"][0] + l3vpn_subinterface = device["interfaces"][1] + lag_interface = copy.deepcopy(physical) + + lag_interface["id"] = "424242" + lag_interface["name"] = "Port-Channel1" + lag_interface["type"] = "LAG" + lag_interface["ip_addresses"] = l3vpn_subinterface["ip_addresses"] + lag_interface["vrf"] = l3vpn_subinterface["vrf"] + lag_interface["lag"] = None + + physical["name"] = "Ethernet1" + physical["lag"] = { + "__typename": "InterfaceType", + "id": lag_interface["id"], + "name": lag_interface["name"], + } + physical["ip_addresses"] = [] + physical["vrf"] = None + device["interfaces"] = [physical, lag_interface] + + serialized = _serialize_router_test_data( + test_data, device, mock_cosmo_config_fixture + ) + + assert "L3VPN" in serialized["routing_instances"] + ri = serialized["routing_instances"]["L3VPN"] + assert ri["interfaces"] == [lag_interface["name"]] + assert ri["instance_type"] == "vrf" + assert serialized["interfaces"][lag_interface["name"]]["type"] == "lag" + assert serialized["interfaces"][lag_interface["name"]]["families"]["inet"] + + +def test_arista_lag_l3vpn_requires_addressing_intent(mock_cosmo_config_fixture): + test_data, device = _arista_l3vpn_test_data() + physical = device["interfaces"][0] + l3vpn_subinterface = device["interfaces"][1] + lag_interface = copy.deepcopy(physical) + lag_interface["id"] = "424242" + lag_interface["name"] = "Port-Channel1" + lag_interface["type"] = "LAG" + lag_interface["ip_addresses"] = [] + lag_interface["vrf"] = l3vpn_subinterface["vrf"] + lag_interface["lag"] = None + lag_interface["tags"] = [] + + physical["name"] = "Ethernet1" + physical["lag"] = { + "__typename": "InterfaceType", + "id": lag_interface["id"], + "name": lag_interface["name"], + } + physical["ip_addresses"] = [] + physical["vrf"] = None + device["interfaces"] = [physical, lag_interface] + + with pytest.raises( + DeviceSerializationError, + match="Port-Channel1.*VRF L3VPN.*no IP address or unnumbered tag", + ): + _serialize_router_test_data( + test_data, + device, + mock_cosmo_config_fixture, + loopbacks=test_data.get("loopbacks", {}), + ) + + +def test_arista_direct_bgp_cpe(mock_cosmo_config_fixture): + test_data = _yaml_load("./test_case_bgpcpe.yml") + device = test_data["device_list"][0] + device["platform"]["manufacturer"]["slug"] = "arista" + device["platform"]["slug"] = "arista-eos64-4-33-8m" + + physical_interface = next( + interface + for interface in device["interfaces"] + if interface["name"] == "ifp-0/1/3" + ) + linked_interface = next( + interface + for interface in device["interfaces"] + if interface["name"] == "ifp-0/1/3.3" + ) + physical_interface.update( + { + "name": "Ethernet3", + "ip_addresses": linked_interface["ip_addresses"], + "mode": linked_interface["mode"], + "tags": linked_interface["tags"], + "vrf": linked_interface["vrf"], + } + ) + device["interfaces"] = [physical_interface] + + serialized = ( + RouterSerializer( + device=device, + l2vpn_list=[], + loopbacks={}, + cosmo_config=mock_cosmo_config_fixture, + ) + .allowPrivateIPs() + .serialize() + ) + groups = serialized["routing_instances"]["default"]["protocols"]["bgp"]["groups"] + assert groups["CPE_Ethernet3_V4"]["peer_as"] == 65086 + assert groups["CPE_Ethernet3_V4"]["neighbors"][0]["peer"] == "10.129.6.12" + + def test_l2vpn_errors(capsys, mock_cosmo_config_fixture): serialize = lambda y: RouterSerializer( device=y["device_list"][0], diff --git a/cosmo/tests/utils.py b/cosmo/tests/utils.py index 555f812e..94ab6c57 100644 --- a/cosmo/tests/utils.py +++ b/cosmo/tests/utils.py @@ -115,7 +115,7 @@ def patchPostFunc(url, json, **kwargs): retVal["interface_list"] = patchKwArgs.get( "connected_devices_interface_list", [] ) - elif 'starts_with: "lo"' in q: + elif 'i_starts_with: "lo"' in q: retVal["interface_list"] = patchKwArgs.get( "loopback_interface_list", [] )