added type hints for emulator and session files

daemon/core/emulator/coreemu.py

@@ -3,13 +3,14 @@ import logging
 import os
 import signal
 import sys
+from typing import Mapping, Type
 
 import core.services
 from core.emulator.session import Session
 from core.services.coreservices import ServiceManager
 
 
-def signal_handler(signal_number, _):
+def signal_handler(signal_number: int, _) -> None:
     """
     Handle signals and force an exit with cleanup.
 
@@ -33,7 +34,7 @@ class CoreEmu:
     Provides logic for creating and configuring CORE sessions and the nodes within them.
     """
 
-    def __init__(self, config=None):
+    def __init__(self, config: Mapping[str, str] = None) -> None:
         """
         Create a CoreEmu object.
 
@@ -57,7 +58,7 @@ class CoreEmu:
         # catch exit event
         atexit.register(self.shutdown)
 
-    def load_services(self):
+    def load_services(self) -> None:
         # load default services
         self.service_errors = core.services.load()
 
@@ -70,7 +71,7 @@ class CoreEmu:
                 custom_service_errors = ServiceManager.add_services(service_path)
                 self.service_errors.extend(custom_service_errors)
 
-    def shutdown(self):
+    def shutdown(self) -> None:
         """
         Shutdown all CORE session.
 
@@ -83,7 +84,7 @@ class CoreEmu:
             session = sessions[_id]
             session.shutdown()
 
-    def create_session(self, _id=None, _cls=Session):
+    def create_session(self, _id: int = None, _cls: Type[Session] = Session) -> Session:
         """
         Create a new CORE session.
 
@@ -101,7 +102,7 @@ class CoreEmu:
         self.sessions[_id] = session
         return session
 
-    def delete_session(self, _id):
+    def delete_session(self, _id: int) -> bool:
         """
         Shutdown and delete a CORE session.
 

daemon/core/emulator/session.py

@@ -12,14 +12,23 @@ import subprocess
 import tempfile
 import threading
 import time
+from typing import Callable, Dict, Iterable, List, Optional, Tuple, Type
 
 from core import constants, utils
 from core.emane.emanemanager import EmaneManager
 from core.emane.nodes import EmaneNet
-from core.emulator.data import EventData, ExceptionData, NodeData
+from core.emulator.data import (
+    ConfigData,
+    EventData,
+    ExceptionData,
+    FileData,
+    LinkData,
+    NodeData,
+)
 from core.emulator.distributed import DistributedController
 from core.emulator.emudata import (
     IdGen,
+    InterfaceData,
     LinkOptions,
     NodeOptions,
     create_interface,
@@ -31,8 +40,9 @@ from core.errors import CoreError
 from core.location.corelocation import CoreLocation
 from core.location.event import EventLoop
 from core.location.mobility import BasicRangeModel, MobilityManager
-from core.nodes.base import CoreNetworkBase, CoreNode, CoreNodeBase
+from core.nodes.base import CoreNetworkBase, CoreNode, CoreNodeBase, NodeBase
 from core.nodes.docker import DockerNode
+from core.nodes.interface import GreTap
 from core.nodes.lxd import LxcNode
 from core.nodes.network import (
     CtrlNet,
@@ -45,7 +55,7 @@ from core.nodes.network import (
 )
 from core.nodes.physical import PhysicalNode, Rj45Node
 from core.plugins.sdt import Sdt
-from core.services.coreservices import CoreServices
+from core.services.coreservices import CoreServices, ServiceBootError
 from core.xml import corexml, corexmldeployment
 from core.xml.corexml import CoreXmlReader, CoreXmlWriter
 
@@ -74,7 +84,9 @@ class Session:
     CORE session manager.
     """
 
-    def __init__(self, _id, config=None, mkdir=True):
+    def __init__(
+        self, _id: int, config: Dict[str, str] = None, mkdir: bool = True
+    ) -> None:
         """
         Create a Session instance.
 
@@ -150,7 +162,7 @@ class Session:
         }
 
     @classmethod
-    def get_node_class(cls, _type):
+    def get_node_class(cls, _type: NodeTypes) -> Type[NodeBase]:
         """
         Retrieve the class for a given node type.
 
@@ -163,20 +175,25 @@ class Session:
         return node_class
 
     @classmethod
-    def get_node_type(cls, _class):
+    def get_node_type(cls, _class: Type[NodeBase]) -> NodeTypes:
         """
         Retrieve node type for a given node class.
 
         :param _class: node class to get a node type for
         :return: node type
         :rtype: core.emulator.enumerations.NodeTypes
+        :raises CoreError: when node type does not exist
         """
         node_type = NODES_TYPE.get(_class)
         if node_type is None:
             raise CoreError(f"invalid node class: {_class}")
         return node_type
 
-    def _link_nodes(self, node_one_id, node_two_id):
+    def _link_nodes(
+        self, node_one_id: int, node_two_id: int
+    ) -> Tuple[
+        CoreNode, CoreNode, CoreNetworkBase, CoreNetworkBase, Tuple[GreTap, GreTap]
+    ]:
         """
         Convenience method for retrieving nodes within link data.
 
@@ -237,14 +254,15 @@ class Session:
         )
         return node_one, node_two, net_one, net_two, tunnel
 
-    def _link_wireless(self, objects, connect):
+    def _link_wireless(self, objects: Iterable[CoreNodeBase], connect: bool) -> None:
         """
         Objects to deal with when connecting/disconnecting wireless links.
 
         :param list objects: possible objects to deal with
         :param bool connect: link interfaces if True, unlink otherwise
         :return: nothing
-        :raises core.CoreError: when objects to link is less than 2, or no common networks are found
+        :raises core.CoreError: when objects to link is less than 2, or no common
+            networks are found
         """
         objects = [x for x in objects if x]
         if len(objects) < 2:
@@ -277,20 +295,23 @@ class Session:
 
     def add_link(
         self,
-        node_one_id,
-        node_two_id,
-        interface_one=None,
-        interface_two=None,
-        link_options=None,
-    ):
+        node_one_id: int,
+        node_two_id: int,
+        interface_one: InterfaceData = None,
+        interface_two: InterfaceData = None,
+        link_options: LinkOptions = None,
+    ) -> None:
         """
         Add a link between nodes.
 
         :param int node_one_id: node one id
         :param int node_two_id: node two id
-        :param core.emulator.emudata.InterfaceData interface_one: node one interface data, defaults to none
-        :param core.emulator.emudata.InterfaceData interface_two: node two interface data, defaults to none
-        :param core.emulator.emudata.LinkOptions link_options: data for creating link, defaults to no options
+        :param core.emulator.emudata.InterfaceData interface_one: node one interface
+            data, defaults to none
+        :param core.emulator.emudata.InterfaceData interface_two: node two interface
+            data, defaults to none
+        :param core.emulator.emudata.LinkOptions link_options: data for creating link,
+            defaults to no options
         :return: nothing
         """
         if not link_options:
@@ -406,12 +427,12 @@ class Session:
 
     def delete_link(
         self,
-        node_one_id,
-        node_two_id,
-        interface_one_id,
-        interface_two_id,
-        link_type=LinkTypes.WIRED,
-    ):
+        node_one_id: int,
+        node_two_id: int,
+        interface_one_id: int,
+        interface_two_id: int,
+        link_type: LinkTypes = LinkTypes.WIRED,
+    ) -> None:
         """
         Delete a link between nodes.
 
@@ -512,12 +533,12 @@ class Session:
 
     def update_link(
         self,
-        node_one_id,
-        node_two_id,
-        interface_one_id=None,
-        interface_two_id=None,
-        link_options=None,
-    ):
+        node_one_id: int,
+        node_two_id: int,
+        interface_one_id: int = None,
+        interface_two_id: int = None,
+        link_options: LinkOptions = None,
+    ) -> None:
         """
         Update link information between nodes.
 
@@ -623,7 +644,13 @@ class Session:
             if node_two:
                 node_two.lock.release()
 
-    def add_node(self, _type=NodeTypes.DEFAULT, _id=None, options=None, _cls=None):
+    def add_node(
+        self,
+        _type: NodeTypes = NodeTypes.DEFAULT,
+        _id: int = None,
+        options: NodeOptions = None,
+        _cls: Type[NodeBase] = None,
+    ) -> NodeBase:
         """
         Add a node to the session, based on the provided node data.
 
@@ -717,14 +744,14 @@ class Session:
 
         return node
 
-    def edit_node(self, node_id, options):
+    def edit_node(self, node_id: int, options: NodeOptions) -> None:
         """
         Edit node information.
 
         :param int node_id: id of node to update
         :param core.emulator.emudata.NodeOptions options: data to update node with
         :return: True if node updated, False otherwise
-        :rtype: bool
+        :rtype: nothing
         :raises core.CoreError: when node to update does not exist
         """
         # get node to update
@@ -737,7 +764,7 @@ class Session:
         node.canvas = options.canvas
         node.icon = options.icon
 
-    def set_node_position(self, node, options):
+    def set_node_position(self, node: NodeBase, options: NodeOptions) -> None:
         """
         Set position for a node, use lat/lon/alt if needed.
 
@@ -767,7 +794,7 @@ class Session:
         if using_lat_lon_alt:
             self.broadcast_node_location(node)
 
-    def broadcast_node_location(self, node):
+    def broadcast_node_location(self, node: NodeBase) -> None:
         """
         Broadcast node location to all listeners.
 
@@ -782,7 +809,7 @@ class Session:
         )
         self.broadcast_node(node_data)
 
-    def start_mobility(self, node_ids=None):
+    def start_mobility(self, node_ids: List[int] = None) -> None:
         """
         Start mobility for the provided node ids.
 
@@ -791,7 +818,7 @@ class Session:
         """
         self.mobility.startup(node_ids)
 
-    def is_active(self):
+    def is_active(self) -> bool:
         """
         Determine if this session is considered to be active. (Runtime or Data collect states)
 
@@ -804,7 +831,7 @@ class Session:
         logging.info("session(%s) checking if active: %s", self.id, result)
         return result
 
-    def open_xml(self, file_name, start=False):
+    def open_xml(self, file_name: str, start: bool = False) -> None:
         """
         Import a session from the EmulationScript XML format.
 
@@ -832,7 +859,7 @@ class Session:
         if start:
             self.instantiate()
 
-    def save_xml(self, file_name):
+    def save_xml(self, file_name: str) -> None:
         """
         Export a session to the EmulationScript XML format.
 
@@ -841,7 +868,7 @@ class Session:
         """
         CoreXmlWriter(self).write(file_name)
 
-    def add_hook(self, state, file_name, source_name, data):
+    def add_hook(self, state: int, file_name: str, source_name: str, data: str) -> None:
         """
         Store a hook from a received file message.
 
@@ -855,7 +882,9 @@ class Session:
         state = f":{state}"
         self.set_hook(state, file_name, source_name, data)
 
-    def add_node_file(self, node_id, source_name, file_name, data):
+    def add_node_file(
+        self, node_id: int, source_name: str, file_name: str, data: str
+    ) -> None:
         """
         Add a file to a node.
 
@@ -873,7 +902,7 @@ class Session:
         elif data is not None:
             node.nodefile(file_name, data)
 
-    def clear(self):
+    def clear(self) -> None:
         """
         Clear all CORE session data. (nodes, hooks, etc)
 
@@ -889,7 +918,7 @@ class Session:
         self.services.reset()
         self.mobility.config_reset()
 
-    def start_events(self):
+    def start_events(self) -> None:
         """
         Start event loop.
 
@@ -897,7 +926,7 @@ class Session:
         """
         self.event_loop.run()
 
-    def mobility_event(self, event_data):
+    def mobility_event(self, event_data: EventData) -> None:
         """
         Handle a mobility event.
 
@@ -906,7 +935,7 @@ class Session:
         """
         self.mobility.handleevent(event_data)
 
-    def set_location(self, lat, lon, alt, scale):
+    def set_location(self, lat: float, lon: float, alt: float, scale: float) -> None:
         """
         Set session geospatial location.
 
@@ -919,7 +948,7 @@ class Session:
         self.location.setrefgeo(lat, lon, alt)
         self.location.refscale = scale
 
-    def shutdown(self):
+    def shutdown(self) -> None:
         """
         Shutdown all session nodes and remove the session directory.
         """
@@ -942,7 +971,7 @@ class Session:
         for handler in self.shutdown_handlers:
             handler(self)
 
-    def broadcast_event(self, event_data):
+    def broadcast_event(self, event_data: EventData) -> None:
         """
         Handle event data that should be provided to event handler.
 
@@ -953,7 +982,7 @@ class Session:
         for handler in self.event_handlers:
             handler(event_data)
 
-    def broadcast_exception(self, exception_data):
+    def broadcast_exception(self, exception_data: ExceptionData) -> None:
         """
         Handle exception data that should be provided to exception handlers.
 
@@ -964,7 +993,7 @@ class Session:
         for handler in self.exception_handlers:
             handler(exception_data)
 
-    def broadcast_node(self, node_data):
+    def broadcast_node(self, node_data: NodeData) -> None:
         """
         Handle node data that should be provided to node handlers.
 
@@ -975,7 +1004,7 @@ class Session:
         for handler in self.node_handlers:
             handler(node_data)
 
-    def broadcast_file(self, file_data):
+    def broadcast_file(self, file_data: FileData) -> None:
         """
         Handle file data that should be provided to file handlers.
 
@@ -986,7 +1015,7 @@ class Session:
         for handler in self.file_handlers:
             handler(file_data)
 
-    def broadcast_config(self, config_data):
+    def broadcast_config(self, config_data: ConfigData) -> None:
         """
         Handle config data that should be provided to config handlers.
 
@@ -997,7 +1026,7 @@ class Session:
         for handler in self.config_handlers:
             handler(config_data)
 
-    def broadcast_link(self, link_data):
+    def broadcast_link(self, link_data: LinkData) -> None:
         """
         Handle link data that should be provided to link handlers.
 
@@ -1008,7 +1037,7 @@ class Session:
         for handler in self.link_handlers:
             handler(link_data)
 
-    def set_state(self, state, send_event=False):
+    def set_state(self, state: EventTypes, send_event: bool = False) -> None:
         """
         Set the session's current state.
 
@@ -1039,7 +1068,7 @@ class Session:
             event_data = EventData(event_type=state_value, time=str(time.monotonic()))
             self.broadcast_event(event_data)
 
-    def write_state(self, state):
+    def write_state(self, state: int) -> None:
         """
         Write the current state to a state file in the session dir.
 
@@ -1053,9 +1082,10 @@ class Session:
         except IOError:
             logging.exception("error writing state file: %s", state)
 
-    def run_hooks(self, state):
+    def run_hooks(self, state: int) -> None:
         """
-        Run hook scripts upon changing states. If hooks is not specified, run all hooks in the given state.
+        Run hook scripts upon changing states. If hooks is not specified, run all hooks
+        in the given state.
 
         :param int state: state to run hooks for
         :return: nothing
@@ -1075,7 +1105,9 @@ class Session:
         else:
             logging.info("no state hooks for %s", state)
 
-    def set_hook(self, hook_type, file_name, source_name, data):
+    def set_hook(
+        self, hook_type: str, file_name: str, source_name: str, data: str
+    ) -> None:
         """
         Store a hook from a received file message.
 
@@ -1107,13 +1139,13 @@ class Session:
             logging.info("immediately running new state hook")
             self.run_hook(hook)
 
-    def del_hooks(self):
+    def del_hooks(self) -> None:
         """
         Clear the hook scripts dict.
         """
         self._hooks.clear()
 
-    def run_hook(self, hook):
+    def run_hook(self, hook: Tuple[str, str]) -> None:
         """
         Run a hook.
 
@@ -1154,7 +1186,7 @@ class Session:
         except (OSError, subprocess.CalledProcessError):
             logging.exception("error running hook: %s", file_name)
 
-    def run_state_hooks(self, state):
+    def run_state_hooks(self, state: int) -> None:
         """
         Run state hooks.
 
@@ -1174,7 +1206,7 @@ class Session:
                     ExceptionLevels.ERROR, "Session.run_state_hooks", None, message
                 )
 
-    def add_state_hook(self, state, hook):
+    def add_state_hook(self, state: int, hook: Callable) -> None:
         """
         Add a state hook.
 
@@ -1190,18 +1222,18 @@ class Session:
         if self.state == state:
             hook(state)
 
-    def del_state_hook(self, state, hook):
+    def del_state_hook(self, state: int, hook: Callable) -> None:
         """
         Delete a state hook.
 
         :param int state: state to delete hook for
         :param func hook: hook to delete
-        :return:
+        :return: nothing
         """
         hooks = self._state_hooks.setdefault(state, [])
         hooks.remove(hook)
 
-    def runtime_state_hook(self, state):
+    def runtime_state_hook(self, state: int) -> None:
         """
         Runtime state hook check.
 
@@ -1217,7 +1249,7 @@ class Session:
             corexmldeployment.CoreXmlDeployment(self, xml_writer.scenario)
             xml_writer.write(xml_file_name)
 
-    def get_environment(self, state=True):
+    def get_environment(self, state: bool = True) -> Dict[str, str]:
         """
         Get an environment suitable for a subprocess.Popen call.
         This is the current process environment with some session-specific
@@ -1265,7 +1297,7 @@ class Session:
 
         return env
 
-    def set_thumbnail(self, thumb_file):
+    def set_thumbnail(self, thumb_file: str) -> None:
         """
         Set the thumbnail filename. Move files from /tmp to session dir.
 
@@ -1281,7 +1313,7 @@ class Session:
         shutil.copy(thumb_file, destination_file)
         self.thumbnail = destination_file
 
-    def set_user(self, user):
+    def set_user(self, user: str) -> None:
         """
         Set the username for this session. Update the permissions of the
         session dir to allow the user write access.
@@ -1299,7 +1331,7 @@ class Session:
 
         self.user = user
 
-    def get_node_id(self):
+    def get_node_id(self) -> int:
         """
         Return a unique, new node id.
         """
@@ -1308,10 +1340,11 @@ class Session:
                 node_id = random.randint(1, 0xFFFF)
                 if node_id not in self.nodes:
                     break
-
         return node_id
 
-    def create_node(self, cls, *args, **kwargs):
+    def create_node(
+        self, cls: Type[NodeBase], *args: Iterable, **kwargs: Dict
+    ) -> NodeBase:
         """
         Create an emulation node.
 
@@ -1322,29 +1355,27 @@ class Session:
         :raises core.CoreError: when id of the node to create already exists
         """
         node = cls(self, *args, **kwargs)
-
         with self._nodes_lock:
             if node.id in self.nodes:
                 node.shutdown()
                 raise CoreError(f"duplicate node id {node.id} for {node.name}")
             self.nodes[node.id] = node
-
         return node
 
-    def get_node(self, _id):
+    def get_node(self, _id: int) -> NodeBase:
         """
         Get a session node.
 
         :param int _id: node id to retrieve
         :return: node for the given id
-        :rtype: core.nodes.base.CoreNode
+        :rtype: core.nodes.base.NodeBase
         :raises core.CoreError: when node does not exist
         """
         if _id not in self.nodes:
             raise CoreError(f"unknown node id {_id}")
         return self.nodes[_id]
 
-    def delete_node(self, _id):
+    def delete_node(self, _id: int) -> bool:
         """
         Delete a node from the session and check if session should shutdown, if no nodes are left.
 
@@ -1365,7 +1396,7 @@ class Session:
 
         return node is not None
 
-    def delete_nodes(self):
+    def delete_nodes(self) -> None:
         """
         Clear the nodes dictionary, and call shutdown for each node.
         """
@@ -1377,7 +1408,7 @@ class Session:
             utils.threadpool(funcs)
         self.node_id_gen.id = 0
 
-    def write_nodes(self):
+    def write_nodes(self) -> None:
         """
         Write nodes to a 'nodes' file in the session dir.
         The 'nodes' file lists: number, name, api-type, class-type
@@ -1392,7 +1423,7 @@ class Session:
         except IOError:
             logging.exception("error writing nodes file")
 
-    def dump_session(self):
+    def dump_session(self) -> None:
         """
         Log information about the session in its current state.
         """
@@ -1405,7 +1436,9 @@ class Session:
             len(self.nodes),
         )
 
-    def exception(self, level, source, node_id, text):
+    def exception(
+        self, level: ExceptionLevels, source: str, node_id: int, text: str
+    ) -> None:
         """
         Generate and broadcast an exception event.
 
@@ -1425,27 +1458,28 @@ class Session:
         )
         self.broadcast_exception(exception_data)
 
-    def instantiate(self):
+    def instantiate(self) -> List[ServiceBootError]:
         """
         We have entered the instantiation state, invoke startup methods
         of various managers and boot the nodes. Validate nodes and check
         for transition to the runtime state.
-        """
 
+        :return: list of service boot errors during startup
+        """
         # write current nodes out to session directory file
         self.write_nodes()
 
         # create control net interfaces and network tunnels
         # which need to exist for emane to sync on location events
         # in distributed scenarios
-        self.add_remove_control_interface(node=None, remove=False)
+        self.add_remove_control_net(0, remove=False)
 
         # initialize distributed tunnels
         self.distributed.start()
 
-        # instantiate will be invoked again upon Emane configure
+        # instantiate will be invoked again upon emane configure
         if self.emane.startup() == self.emane.NOT_READY:
-            return
+            return []
 
         # boot node services and then start mobility
         exceptions = self.boot_nodes()
@@ -1462,12 +1496,13 @@ class Session:
             self.check_runtime()
         return exceptions
 
-    def get_node_count(self):
+    def get_node_count(self) -> int:
         """
         Returns the number of CoreNodes and CoreNets, except for those
         that are not considered in the GUI's node count.
-        """
 
+        :return: created node count
+        """
         with self._nodes_lock:
             count = 0
             for node_id in self.nodes:
@@ -1480,14 +1515,15 @@ class Session:
                     continue
 
                 count += 1
-
         return count
 
-    def check_runtime(self):
+    def check_runtime(self) -> None:
         """
         Check if we have entered the runtime state, that all nodes have been
         started and the emulation is running. Start the event loop once we
         have entered runtime (time=0).
+
+        :return: nothing
         """
         # this is called from instantiate() after receiving an event message
         # for the instantiation state
@@ -1504,10 +1540,12 @@ class Session:
         self.event_loop.run()
         self.set_state(EventTypes.RUNTIME_STATE, send_event=True)
 
-    def data_collect(self):
+    def data_collect(self) -> None:
         """
         Tear down a running session. Stop the event loop and any running
         nodes, and perform clean-up.
+
+        :return: nothing
         """
         # stop event loop
         self.event_loop.stop()
@@ -1528,51 +1566,52 @@ class Session:
         # update control interface hosts
         self.update_control_interface_hosts(remove=True)
 
-        # remove all four possible control networks. Does nothing if ctrlnet is not
-        # installed.
-        self.add_remove_control_interface(node=None, net_index=0, remove=True)
-        self.add_remove_control_interface(node=None, net_index=1, remove=True)
-        self.add_remove_control_interface(node=None, net_index=2, remove=True)
-        self.add_remove_control_interface(node=None, net_index=3, remove=True)
+        # remove all four possible control networks
+        self.add_remove_control_net(0, remove=True)
+        self.add_remove_control_net(1, remove=True)
+        self.add_remove_control_net(2, remove=True)
+        self.add_remove_control_net(3, remove=True)
 
-    def check_shutdown(self):
+    def check_shutdown(self) -> bool:
         """
         Check if we have entered the shutdown state, when no running nodes
         and links remain.
+
+        :return: True if should shutdown, False otherwise
         """
         node_count = self.get_node_count()
         logging.debug(
             "session(%s) checking shutdown: %s nodes remaining", self.id, node_count
         )
-
         shutdown = False
         if node_count == 0:
             shutdown = True
             self.set_state(EventTypes.SHUTDOWN_STATE)
-
         return shutdown
 
-    def short_session_id(self):
+    def short_session_id(self) -> str:
         """
         Return a shorter version of the session ID, appropriate for
         interface names, where length may be limited.
+
+        :return: short session id
         """
         ssid = (self.id >> 8) ^ (self.id & ((1 << 8) - 1))
         return f"{ssid:x}"
 
-    def boot_node(self, node):
+    def boot_node(self, node: CoreNode) -> None:
         """
         Boot node by adding a control interface when necessary and starting
         node services.
 
-        :param core.nodes.base.CoreNodeBase node: node to boot
+        :param core.nodes.base.CoreNode node: node to boot
         :return: nothing
         """
         logging.info("booting node(%s): %s", node.name, [x.name for x in node.services])
         self.add_remove_control_interface(node=node, remove=False)
         self.services.boot_services(node)
 
-    def boot_nodes(self):
+    def boot_nodes(self) -> List[ServiceBootError]:
         """
         Invoke the boot() procedure for all nodes and send back node
         messages to the GUI for node messages that had the status
@@ -1596,7 +1635,7 @@ class Session:
             self.update_control_interface_hosts()
         return exceptions
 
-    def get_control_net_prefixes(self):
+    def get_control_net_prefixes(self) -> List[str]:
         """
         Retrieve control net prefixes.
 
@@ -1608,13 +1647,11 @@ class Session:
         p1 = self.options.get_config("controlnet1")
         p2 = self.options.get_config("controlnet2")
         p3 = self.options.get_config("controlnet3")
-
         if not p0 and p:
             p0 = p
-
         return [p0, p1, p2, p3]
 
-    def get_control_net_server_interfaces(self):
+    def get_control_net_server_interfaces(self) -> List[str]:
         """
         Retrieve control net server interfaces.
 
@@ -1629,7 +1666,7 @@ class Session:
         d3 = self.options.get_config("controlnetif3")
         return [None, d1, d2, d3]
 
-    def get_control_net_index(self, dev):
+    def get_control_net_index(self, dev: str) -> int:
         """
         Retrieve control net index.
 
@@ -1645,10 +1682,22 @@ class Session:
                 return index
         return -1
 
-    def get_control_net(self, net_index):
-        return self.get_node(CTRL_NET_ID + net_index)
+    def get_control_net(self, net_index: int) -> CtrlNet:
+        """
+        Retrieve a control net based on index.
 
-    def add_remove_control_net(self, net_index, remove=False, conf_required=True):
+        :param net_index: control net index
+        :return: control net
+        :raises CoreError: when control net is not found
+        """
+        node = self.get_node(CTRL_NET_ID + net_index)
+        if not isinstance(node, CtrlNet):
+            raise CoreError("node is not a valid CtrlNet: %s", node.name)
+        return node
+
+    def add_remove_control_net(
+        self, net_index: int, remove: bool = False, conf_required: bool = True
+    ) -> Optional[CtrlNet]:
         """
         Create a control network bridge as necessary.
         When the remove flag is True, remove the bridge that connects control
@@ -1682,11 +1731,9 @@ class Session:
         # return any existing controlnet bridge
         try:
             control_net = self.get_control_net(net_index)
-
             if remove:
                 self.delete_node(control_net.id)
                 return None
-
             return control_net
         except CoreError:
             if remove:
@@ -1730,12 +1777,15 @@ class Session:
             updown_script=updown_script,
             serverintf=server_interface,
         )
-
         return control_net
 
     def add_remove_control_interface(
-        self, node, net_index=0, remove=False, conf_required=True
-    ):
+        self,
+        node: CoreNode,
+        net_index: int = 0,
+        remove: bool = False,
+        conf_required: bool = True,
+    ) -> None:
         """
         Add a control interface to a node when a 'controlnet' prefix is
         listed in the config file or session options. Uses
@@ -1782,7 +1832,9 @@ class Session:
         )
         node.netif(interface1).control = True
 
-    def update_control_interface_hosts(self, net_index=0, remove=False):
+    def update_control_interface_hosts(
+        self, net_index: int = 0, remove: bool = False
+    ) -> None:
         """
         Add the IP addresses of control interfaces to the /etc/hosts file.
 
@@ -1813,10 +1865,9 @@ class Session:
                 entries.append(f"{address} {name}")
 
         logging.info("Adding %d /etc/hosts file entries.", len(entries))
-
         utils.file_munge("/etc/hosts", header, "\n".join(entries) + "\n")
 
-    def runtime(self):
+    def runtime(self) -> float:
         """
         Return the current time we have been in the runtime state, or zero
         if not in runtime.
@@ -1826,7 +1877,13 @@ class Session:
         else:
             return 0.0
 
-    def add_event(self, event_time, node=None, name=None, data=None):
+    def add_event(
+        self,
+        event_time: float,
+        node: CoreNode = None,
+        name: str = None,
+        data: str = None,
+    ) -> None:
         """
         Add an event to the event queue, with a start time relative to the
         start of the runtime state.
@@ -1865,7 +1922,9 @@ class Session:
 
     # TODO: if data is None, this blows up, but this ties into how event functions
     #  are ran, need to clean that up
-    def run_event(self, node_id=None, name=None, data=None):
+    def run_event(
+        self, node_id: int = None, name: str = None, data: str = None
+    ) -> None:
         """
         Run a scheduled event, executing commands in the data string.
 

daemon/core/nodes/base.py

@@ -948,6 +948,29 @@ class CoreNetworkBase(NodeBase):
         """
         raise NotImplementedError
 
+    def linknet(self, net):
+        """
+        Link network to another.
+
+        :param core.nodes.base.CoreNetworkBase net: network to link with
+        :return: created interface
+        :rtype: core.nodes.interface.Veth
+        """
+        pass
+
+    def getlinknetif(self, net):
+        """
+        Return the interface of that links this net with another net.
+
+        :param core.nodes.base.CoreNetworkBase net: interface to get link for
+        :return: interface the provided network is linked to
+        :rtype: core.nodes.interface.CoreInterface
+        """
+        for netif in self.netifs():
+            if hasattr(netif, "othernet") and netif.othernet == net:
+                return netif
+        return None
+
     def attach(self, netif):
         """
         Attach network interface.

daemon/core/nodes/network.py

@@ -525,9 +525,9 @@ class CoreNetwork(CoreNetworkBase):
         Link this bridge with another by creating a veth pair and installing
         each device into each bridge.
 
-        :param core.netns.vnet.LxBrNet net: network to link with
+        :param core.nodes.base.CoreNetworkBase net: network to link with
         :return: created interface
-        :rtype: Veth
+        :rtype: core.nodes.interface.Veth
         """
         sessionid = self.session.short_session_id()
         try: