335 lines
9 KiB
Python
335 lines
9 KiB
Python
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)
|