diff --git a/daemon/core/nodes/base.py b/daemon/core/nodes/base.py index 92717ead..f5688f5a 100644 --- a/daemon/core/nodes/base.py +++ b/daemon/core/nodes/base.py @@ -9,7 +9,7 @@ import threading from dataclasses import dataclass, field from pathlib import Path from threading import RLock -from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Type, Union +from typing import TYPE_CHECKING, Optional, Union import netaddr @@ -29,10 +29,10 @@ if TYPE_CHECKING: from core.configservice.base import ConfigService from core.services.coreservices import CoreService - CoreServices = List[Union[CoreService, Type[CoreService]]] - ConfigServiceType = Type[ConfigService] + CoreServices = list[Union[CoreService, type[CoreService]]] + ConfigServiceType = type[ConfigService] -PRIVATE_DIRS: List[Path] = [Path("/var/run"), Path("/var/log")] +PRIVATE_DIRS: list[Path] = [Path("/var/run"), Path("/var/log")] @dataclass @@ -64,7 +64,7 @@ class Position: self.z = z return True - def get(self) -> Tuple[float, float, float]: + def get(self) -> tuple[float, float, float]: """ Retrieve x,y,z position. @@ -88,7 +88,7 @@ class Position: self.lat = lat self.alt = alt - def get_geo(self) -> Tuple[float, float, float]: + def get_geo(self) -> tuple[float, float, float]: """ Retrieve current geo position lon, lat, alt. @@ -113,9 +113,9 @@ class NodeOptions: class CoreNodeOptions(NodeOptions): model: str = "PC" """model is used for providing a default set of services""" - services: List[str] = field(default_factory=list) + services: list[str] = field(default_factory=list) """services to start within node""" - config_services: List[str] = field(default_factory=list) + config_services: list[str] = field(default_factory=list) """config services to start within node""" directory: Path = None """directory to define node, defaults to path under the session directory""" @@ -152,7 +152,7 @@ class NodeBase(abc.ABC): self.server: "DistributedServer" = server self.model: Optional[str] = None self.services: CoreServices = [] - self.ifaces: Dict[int, CoreInterface] = {} + self.ifaces: dict[int, CoreInterface] = {} self.iface_id: int = 0 self.position: Position = Position() self.up: bool = False @@ -201,7 +201,7 @@ class NodeBase(abc.ABC): def host_cmd( self, args: str, - env: Dict[str, str] = None, + env: dict[str, str] = None, cwd: Path = None, wait: bool = True, shell: bool = False, @@ -246,7 +246,7 @@ class NodeBase(abc.ABC): """ return self.position.set(x=x, y=y, z=z) - def getposition(self) -> Tuple[float, float, float]: + def getposition(self) -> tuple[float, float, float]: """ Return an (x,y,z) tuple representing this object's position. @@ -331,7 +331,7 @@ class NodeBase(abc.ABC): raise CoreError(f"node({self.name}) does not have interface({iface_id})") return self.ifaces[iface_id] - def get_ifaces(self, control: bool = True) -> List[CoreInterface]: + def get_ifaces(self, control: bool = True) -> list[CoreInterface]: """ Retrieve sorted list of interfaces, optionally do not include control interfaces. @@ -395,7 +395,7 @@ class CoreNodeBase(NodeBase): will run on, default is None for localhost """ super().__init__(session, _id, name, server, options) - self.config_services: Dict[str, "ConfigService"] = {} + self.config_services: dict[str, "ConfigService"] = {} self.directory: Optional[Path] = None self.tmpnodedir: bool = False @@ -481,7 +481,7 @@ class CoreNodeBase(NodeBase): raise CoreError(f"node({self.name}) already has service({name})") self.config_services[name] = service_class(self) - def set_service_config(self, name: str, data: Dict[str, str]) -> None: + def set_service_config(self, name: str, data: dict[str, str]) -> None: """ Sets configuration service custom config data. @@ -574,7 +574,7 @@ class CoreNode(CoreNodeBase): self.directory: Optional[Path] = options.directory self.ctrlchnlname: Path = self.session.directory / self.name self.pid: Optional[int] = None - self._mounts: List[Tuple[Path, Path]] = [] + self._mounts: list[tuple[Path, Path]] = [] self.node_net_client: LinuxNetClient = self.create_node_net_client( self.session.use_ovs() ) @@ -928,7 +928,7 @@ class CoreNetworkBase(NodeBase): mtu = self.session.options.get_int("mtu") self.mtu: int = mtu if mtu > 0 else DEFAULT_MTU self.brname: Optional[str] = None - self.linked: Dict[CoreInterface, Dict[CoreInterface, bool]] = {} + self.linked: dict[CoreInterface, dict[CoreInterface, bool]] = {} self.linked_lock: threading.Lock = threading.Lock() def attach(self, iface: CoreInterface) -> None: diff --git a/daemon/core/nodes/docker.py b/daemon/core/nodes/docker.py index 45d2b892..ba470167 100644 --- a/daemon/core/nodes/docker.py +++ b/daemon/core/nodes/docker.py @@ -4,7 +4,7 @@ import shlex from dataclasses import dataclass, field from pathlib import Path from tempfile import NamedTemporaryFile -from typing import TYPE_CHECKING, Dict, List, Tuple +from typing import TYPE_CHECKING from core.emulator.distributed import DistributedServer from core.errors import CoreCommandError, CoreError @@ -23,9 +23,9 @@ DOCKER: str = "docker" class DockerOptions(CoreNodeOptions): image: str = "ubuntu" """image used when creating container""" - binds: List[Tuple[str, str]] = field(default_factory=list) + binds: list[tuple[str, str]] = field(default_factory=list) """bind mount source and destinations to setup within container""" - volumes: List[Tuple[str, str, bool, bool]] = field(default_factory=list) + volumes: list[tuple[str, str, bool, bool]] = field(default_factory=list) """ volume mount source, destination, unique, delete to setup within container @@ -74,8 +74,8 @@ class DockerNode(CoreNode): options = options or DockerOptions() super().__init__(session, _id, name, server, options) self.image: str = options.image - self.binds: List[Tuple[str, str]] = options.binds - self.volumes: Dict[str, DockerVolume] = {} + self.binds: list[tuple[str, str]] = options.binds + self.volumes: dict[str, DockerVolume] = {} for src, dst, unique, delete in options.volumes: src_name = self._unique_name(src) if unique else src self.volumes[src] = DockerVolume(src_name, dst, unique, delete) diff --git a/daemon/core/nodes/interface.py b/daemon/core/nodes/interface.py index bb90653f..294e85f9 100644 --- a/daemon/core/nodes/interface.py +++ b/daemon/core/nodes/interface.py @@ -5,7 +5,7 @@ virtual ethernet classes that implement the interfaces available under Linux. import logging import math from pathlib import Path -from typing import TYPE_CHECKING, Callable, Dict, List, Optional +from typing import TYPE_CHECKING, Callable, Optional import netaddr @@ -114,8 +114,8 @@ class CoreInterface: self.up: bool = False self.mtu: int = mtu self.net: Optional[CoreNetworkBase] = None - self.ip4s: List[netaddr.IPNetwork] = [] - self.ip6s: List[netaddr.IPNetwork] = [] + self.ip4s: list[netaddr.IPNetwork] = [] + self.ip6s: list[netaddr.IPNetwork] = [] self.mac: Optional[netaddr.EUI] = None # placeholder position hook self.poshook: Callable[[CoreInterface], None] = lambda x: None @@ -133,7 +133,7 @@ class CoreInterface: def host_cmd( self, args: str, - env: Dict[str, str] = None, + env: dict[str, str] = None, cwd: Path = None, wait: bool = True, shell: bool = False, @@ -235,7 +235,7 @@ class CoreInterface: """ return next(iter(self.ip6s), None) - def ips(self) -> List[netaddr.IPNetwork]: + def ips(self) -> list[netaddr.IPNetwork]: """ Retrieve a list of all ip4 and ip6 addresses combined. diff --git a/daemon/core/nodes/lxd.py b/daemon/core/nodes/lxd.py index 01bd2db7..497c228e 100644 --- a/daemon/core/nodes/lxd.py +++ b/daemon/core/nodes/lxd.py @@ -5,7 +5,7 @@ import time from dataclasses import dataclass, field from pathlib import Path from tempfile import NamedTemporaryFile -from typing import TYPE_CHECKING, Dict, List, Tuple +from typing import TYPE_CHECKING from core.emulator.data import InterfaceData, LinkOptions from core.emulator.distributed import DistributedServer @@ -24,9 +24,9 @@ if TYPE_CHECKING: class LxcOptions(CoreNodeOptions): image: str = "ubuntu" """image used when creating container""" - binds: List[Tuple[str, str]] = field(default_factory=list) + binds: list[tuple[str, str]] = field(default_factory=list) """bind mount source and destinations to setup within container""" - volumes: List[Tuple[str, str, bool, bool]] = field(default_factory=list) + volumes: list[tuple[str, str, bool, bool]] = field(default_factory=list) """ volume mount source, destination, unique, delete to setup within container @@ -74,7 +74,7 @@ class LxcNode(CoreNode): args = f"{BASH} -c {shlex.quote(args)}" return f"nsenter -t {self.pid} -m -u -i -p -n {args}" - def _get_info(self) -> Dict: + def _get_info(self) -> dict: args = f"lxc list {self.name} --format json" output = self.host_cmd(args) data = json.loads(output) diff --git a/daemon/core/nodes/network.py b/daemon/core/nodes/network.py index e201e807..1ea9c31e 100644 --- a/daemon/core/nodes/network.py +++ b/daemon/core/nodes/network.py @@ -6,7 +6,7 @@ import logging import threading from dataclasses import dataclass from pathlib import Path -from typing import TYPE_CHECKING, Dict, List, Optional, Type +from typing import TYPE_CHECKING, Optional import netaddr @@ -51,7 +51,7 @@ class NftablesQueue: # this lock protects cmds and updates lists self.lock: threading.Lock = threading.Lock() # list of pending nftables commands - self.cmds: List[str] = [] + self.cmds: list[str] = [] # list of WLANs requiring update self.updates: utils.SetQueue = utils.SetQueue() @@ -226,7 +226,7 @@ class CoreNetwork(CoreNetworkBase): def host_cmd( self, args: str, - env: Dict[str, str] = None, + env: dict[str, str] = None, cwd: Path = None, wait: bool = True, shell: bool = False, @@ -448,7 +448,7 @@ class GreTapBridge(CoreNetwork): self.gretap = None super().shutdown() - def add_ips(self, ips: List[str]) -> None: + def add_ips(self, ips: list[str]) -> None: """ Set the remote tunnel endpoint. This is a one-time method for creating the GreTap device, which requires the remoteip at startup. @@ -512,7 +512,7 @@ class CtrlNet(CoreNetwork): policy: NetworkPolicy = NetworkPolicy.ACCEPT # base control interface index CTRLIF_IDX_BASE: int = 99 - DEFAULT_PREFIX_LIST: List[str] = [ + DEFAULT_PREFIX_LIST: list[str] = [ "172.16.0.0/24 172.16.1.0/24 172.16.2.0/24 172.16.3.0/24 172.16.4.0/24", "172.17.0.0/24 172.17.1.0/24 172.17.2.0/24 172.17.3.0/24 172.17.4.0/24", "172.18.0.0/24 172.18.1.0/24 172.18.2.0/24 172.18.3.0/24 172.18.4.0/24", @@ -734,7 +734,7 @@ class WlanNode(CoreNetwork): iface.poshook = self.wireless_model.position_callback iface.setposition() - def setmodel(self, wireless_model: Type["WirelessModel"], config: Dict[str, str]): + def setmodel(self, wireless_model: type["WirelessModel"], config: dict[str, str]): """ Sets the mobility and wireless model. @@ -753,12 +753,12 @@ class WlanNode(CoreNetwork): self.mobility = wireless_model(session=self.session, _id=self.id) self.mobility.update_config(config) - def update_mobility(self, config: Dict[str, str]) -> None: + def update_mobility(self, config: dict[str, str]) -> None: if not self.mobility: raise CoreError(f"no mobility set to update for node({self.name})") self.mobility.update_config(config) - def updatemodel(self, config: Dict[str, str]) -> None: + def updatemodel(self, config: dict[str, str]) -> None: if not self.wireless_model: raise CoreError(f"no model set to update for node({self.name})") logger.debug( @@ -768,7 +768,7 @@ class WlanNode(CoreNetwork): for iface in self.get_ifaces(): iface.setposition() - def links(self, flags: MessageFlags = MessageFlags.NONE) -> List[LinkData]: + def links(self, flags: MessageFlags = MessageFlags.NONE) -> list[LinkData]: """ Retrieve all link data. diff --git a/daemon/core/nodes/physical.py b/daemon/core/nodes/physical.py index 8ab13f20..037ef901 100644 --- a/daemon/core/nodes/physical.py +++ b/daemon/core/nodes/physical.py @@ -4,7 +4,7 @@ PhysicalNode class for including real systems in the emulated network. import logging from pathlib import Path -from typing import TYPE_CHECKING, List, Optional, Tuple +from typing import TYPE_CHECKING, Optional import netaddr @@ -52,7 +52,7 @@ class Rj45Node(CoreNodeBase): ) self.iface.transport_type = TransportType.RAW self.old_up: bool = False - self.old_addrs: List[Tuple[str, Optional[str]]] = [] + self.old_addrs: list[tuple[str, Optional[str]]] = [] def startup(self) -> None: """ @@ -159,7 +159,7 @@ class Rj45Node(CoreNodeBase): """ # TODO: save/restore the PROMISC flag self.old_up = False - self.old_addrs: List[Tuple[str, Optional[str]]] = [] + self.old_addrs: list[tuple[str, Optional[str]]] = [] localname = self.iface.localname output = self.net_client.address_show(localname) for line in output.split("\n"): diff --git a/daemon/core/nodes/wireless.py b/daemon/core/nodes/wireless.py index ef37db35..51a98917 100644 --- a/daemon/core/nodes/wireless.py +++ b/daemon/core/nodes/wireless.py @@ -7,7 +7,7 @@ import logging import math import secrets from dataclasses import dataclass -from typing import TYPE_CHECKING, Dict, List, Set, Tuple +from typing import TYPE_CHECKING from core.config import ConfigBool, ConfigFloat, ConfigInt, Configuration from core.emulator.data import LinkData, LinkOptions @@ -41,7 +41,7 @@ KEY_LOSS: str = "loss" def calc_distance( - point1: Tuple[float, float, float], point2: Tuple[float, float, float] + point1: tuple[float, float, float], point2: tuple[float, float, float] ) -> float: a = point1[0] - point2[0] b = point1[1] - point2[1] @@ -51,7 +51,7 @@ def calc_distance( return math.hypot(math.hypot(a, b), c) -def get_key(node1_id: int, node2_id: int) -> Tuple[int, int]: +def get_key(node1_id: int, node2_id: int) -> tuple[int, int]: return (node1_id, node2_id) if node1_id < node2_id else (node2_id, node1_id) @@ -65,7 +65,7 @@ class WirelessLink: class WirelessNode(CoreNetworkBase): - options: List[Configuration] = [ + options: list[Configuration] = [ ConfigBool( id=KEY_ENABLED, default="1" if CONFIG_ENABLED else "0", label="Enabled?" ), @@ -87,7 +87,7 @@ class WirelessNode(CoreNetworkBase): ), ConfigFloat(id=KEY_LOSS, default=str(CONFIG_LOSS), label="Loss Initial"), ] - devices: Set[str] = set() + devices: set[str] = set() @classmethod def add_device(cls) -> str: @@ -111,8 +111,8 @@ class WirelessNode(CoreNetworkBase): options: NodeOptions = None, ): super().__init__(session, _id, name, server, options) - self.bridges: Dict[int, Tuple[CoreInterface, str]] = {} - self.links: Dict[Tuple[int, int], WirelessLink] = {} + self.bridges: dict[int, tuple[CoreInterface, str]] = {} + self.links: dict[tuple[int, int], WirelessLink] = {} self.position_enabled: bool = CONFIG_ENABLED self.bandwidth: int = CONFIG_BANDWIDTH self.delay: int = CONFIG_DELAY @@ -321,7 +321,7 @@ class WirelessNode(CoreNetworkBase): def adopt_iface(self, iface: CoreInterface, name: str) -> None: raise CoreError(f"{type(self)} does not support adopt interface") - def get_config(self) -> Dict[str, Configuration]: + def get_config(self) -> dict[str, Configuration]: config = {x.id: x for x in copy.copy(self.options)} config[KEY_ENABLED].default = "1" if self.position_enabled else "0" config[KEY_RANGE].default = str(self.max_range) @@ -333,7 +333,7 @@ class WirelessNode(CoreNetworkBase): config[KEY_JITTER].default = str(self.jitter) return config - def set_config(self, config: Dict[str, str]) -> None: + def set_config(self, config: dict[str, str]) -> None: logger.info("wireless config: %s", config) self.position_enabled = config[KEY_ENABLED] == "1" self.max_range = float(config[KEY_RANGE])