from typing import List, Optional import netaddr from core import utils from core.api.grpc.core_pb2 import LinkOptions from core.emane.nodes import EmaneNet from core.emulator.enumerations import LinkTypes from core.nodes.base import CoreNetworkBase, CoreNode from core.nodes.interface import CoreInterface from core.nodes.physical import PhysicalNode class IdGen: def __init__(self, _id: int = 0) -> None: self.id = _id def next(self) -> int: self.id += 1 return self.id def link_config( network: CoreNetworkBase, interface: CoreInterface, link_options: LinkOptions, devname: str = None, interface_two: CoreInterface = None, ) -> None: """ Convenience method for configuring a link, :param network: network to configure link for :param interface: interface to configure :param link_options: data to configure link with :param devname: device name, default is None :param interface_two: other interface associated, default is None :return: nothing """ config = { "netif": interface, "bw": link_options.bandwidth, "delay": link_options.delay, "loss": link_options.per, "duplicate": link_options.dup, "jitter": link_options.jitter, "netif2": interface_two, } # hacky check here, because physical and emane nodes do not conform to the same # linkconfig interface if not isinstance(network, (EmaneNet, PhysicalNode)): config["devname"] = devname network.linkconfig(**config) class NodeOptions: """ Options for creating and updating nodes within core. """ def __init__(self, name: str = None, model: str = "PC", image: str = None) -> None: """ Create a NodeOptions object. :param name: name of node, defaults to node class name postfix with its id :param model: defines services for default and physical nodes, defaults to "router" :param image: image to use for docker nodes """ self.name = name self.model = model self.canvas = None self.icon = None self.opaque = None self.services = [] self.config_services = [] self.x = None self.y = None self.lat = None self.lon = None self.alt = None self.emulation_id = None self.server = None self.image = image self.emane = None def set_position(self, x: float, y: float) -> None: """ Convenience method for setting position. :param x: x position :param y: y position :return: nothing """ self.x = x self.y = y def set_location(self, lat: float, lon: float, alt: float) -> None: """ Convenience method for setting location. :param lat: latitude :param lon: longitude :param alt: altitude :return: nothing """ self.lat = lat self.lon = lon self.alt = alt class LinkOptions: """ Options for creating and updating links within core. """ def __init__(self, _type: LinkTypes = LinkTypes.WIRED) -> None: """ Create a LinkOptions object. :param _type: type of link, defaults to wired """ self.type = _type self.session = None self.delay = None self.bandwidth = None self.per = None self.dup = None self.jitter = None self.mer = None self.burst = None self.mburst = None self.gui_attributes = None self.unidirectional = None self.emulation_id = None self.network_id = None self.key = None self.opaque = None class InterfaceData: """ Convenience class for storing interface data. """ def __init__( self, _id: int, name: str, mac: str, ip4: str, ip4_mask: int, ip6: str, ip6_mask: int, ) -> None: """ Creates an InterfaceData object. :param _id: interface id :param name: name for interface :param mac: mac address :param ip4: ipv4 address :param ip4_mask: ipv4 bit mask :param ip6: ipv6 address :param ip6_mask: ipv6 bit mask """ self.id = _id self.name = name self.mac = mac self.ip4 = ip4 self.ip4_mask = ip4_mask self.ip6 = ip6 self.ip6_mask = ip6_mask def has_ip4(self) -> bool: """ Determines if interface has an ip4 address. :return: True if has ip4, False otherwise """ return all([self.ip4, self.ip4_mask]) def has_ip6(self) -> bool: """ Determines if interface has an ip6 address. :return: True if has ip6, False otherwise """ return all([self.ip6, self.ip6_mask]) def ip4_address(self) -> Optional[str]: """ Retrieve a string representation of the ip4 address and netmask. :return: ip4 string or None """ if self.has_ip4(): return f"{self.ip4}/{self.ip4_mask}" else: return None def ip6_address(self) -> Optional[str]: """ Retrieve a string representation of the ip6 address and netmask. :return: ip4 string or None """ if self.has_ip6(): return f"{self.ip6}/{self.ip6_mask}" else: return None def get_addresses(self) -> List[str]: """ Returns a list of ip4 and ip6 address when present. :return: list of addresses """ ip4 = self.ip4_address() ip6 = self.ip6_address() return [i for i in [ip4, ip6] if i] class IpPrefixes: """ Convenience class to help generate IP4 and IP6 addresses for nodes within CORE. """ def __init__(self, ip4_prefix: str = None, ip6_prefix: str = None) -> None: """ Creates an IpPrefixes object. :param ip4_prefix: ip4 prefix to use for generation :param ip6_prefix: ip6 prefix to use for generation :raises ValueError: when both ip4 and ip6 prefixes have not been provided """ if not ip4_prefix and not ip6_prefix: raise ValueError("ip4 or ip6 must be provided") self.ip4 = None if ip4_prefix: self.ip4 = netaddr.IPNetwork(ip4_prefix) self.ip6 = None if ip6_prefix: self.ip6 = netaddr.IPNetwork(ip6_prefix) def ip4_address(self, node: CoreNode) -> str: """ Convenience method to return the IP4 address for a node. :param node: node to get IP4 address for :return: IP4 address or None """ if not self.ip4: raise ValueError("ip4 prefixes have not been set") return str(self.ip4[node.id]) def ip6_address(self, node: CoreNode) -> str: """ Convenience method to return the IP6 address for a node. :param node: node to get IP6 address for :return: IP4 address or None """ if not self.ip6: raise ValueError("ip6 prefixes have not been set") return str(self.ip6[node.id]) def create_interface( self, node: CoreNode, name: str = None, mac: str = None ) -> InterfaceData: """ Creates interface data for linking nodes, using the nodes unique id for generation, along with a random mac address, unless provided. :param node: node to create interface for :param name: name to set for interface, default is eth{id} :param mac: mac address to use for this interface, default is random generation :return: new interface data for the provided node """ # interface id inteface_id = node.newifindex() # generate ip4 data ip4 = None ip4_mask = None if self.ip4: ip4 = self.ip4_address(node) ip4_mask = self.ip4.prefixlen # generate ip6 data ip6 = None ip6_mask = None if self.ip6: ip6 = self.ip6_address(node) ip6_mask = self.ip6.prefixlen # random mac if not mac: mac = utils.random_mac() return InterfaceData( _id=inteface_id, name=name, ip4=ip4, ip4_mask=ip4_mask, ip6=ip6, ip6_mask=ip6_mask, mac=mac, ) def create_interface( node: CoreNode, network: CoreNetworkBase, interface_data: InterfaceData ): """ Create an interface for a node on a network using provided interface data. :param node: node to create interface for :param network: network to associate interface with :param interface_data: interface data :return: created interface """ node.newnetif( network, addrlist=interface_data.get_addresses(), hwaddr=interface_data.mac, ifindex=interface_data.id, ifname=interface_data.name, ) return node.netif(interface_data.id)