gns3-server/gns3server/compute/base_node.py

#
# Copyright (C) 2015 GNS3 Technologies Inc.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

import sys
import os
import stat
import shutil
import asyncio
import contextlib
import json
import struct
import tempfile
import psutil
import platform
import re
import asyncssh

from fastapi import WebSocketDisconnect
from gns3server.utils.interfaces import interfaces
from gns3server.compute.compute_error import ComputeError
from ..compute.port_manager import PortManager
from ..utils.asyncio import wait_run_in_executor, locking
from ..utils.asyncio.ssh_server import AsyncioSSHServer
from ..utils.asyncio.telnet_server import AsyncioTelnetServer
from gns3server.compute.ubridge.hypervisor import Hypervisor
from gns3server.compute.ubridge.ubridge_error import UbridgeError
from .nios.nio_udp import NIOUDP
from .error import NodeError

import logging

log = logging.getLogger(__name__)


class BaseNode:

    """
    Base node implementation.

    :param name: name of this node
    :param node_id: Node instance identifier
    :param project: Project instance
    :param manager: parent node manager
    :param console: console TCP port
    :param console_type: console type
    :param aux: auxiliary console TCP port
    :param aux_type: auxiliary console type
    :param linked_clone: The node base image is duplicate/overlay (Each node data are independent)
    :param wrap_console: The console is wrapped using a proxy transport server
    :param wrap_aux: The auxiliary console is wrapped using a proxy transport server
    """

    def __init__(
        self,
        name,
        node_id,
        project,
        manager,
        console=None,
        console_type="telnet",
        aux=None,
        aux_type="none",
        linked_clone=True,
        wrap_console=False,
        wrap_aux=False,
    ):

        self._name = name
        self._usage = ""
        self._id = node_id
        self._linked_clone = linked_clone
        self._project = project
        self._manager = manager
        self._console = console
        self._aux = aux
        self._console_type = console_type
        self._aux_type = aux_type
        self._temporary_directory = None
        self._hw_virtualization = False
        self._ubridge_hypervisor = None
        self._closed = False
        self._node_status = "stopped"
        self._command_line = ""
        self._wrap_console = wrap_console
        self._wrap_aux = wrap_aux
        self._wrapper_telnet_servers = []
        self._wrap_console_reader = None
        self._wrap_console_writer = None
        self._internal_console_port = None
        self._internal_aux_port = None
        self._custom_adapters = []
        self._ubridge_require_privileged_access = False
        # marker filter name -> uBridge bridge_name (recorded at apply time so
        # _ubridge_set_marker_filter_state can toggle on/off without an NIO rebuild).
        self._marker_filter_bridges = {}
        # Parallel store of the installed marker spec (bpf/tag/direction/enabled/...)
        # keyed by (name, link_id) so _ubridge_apply_markers can reconcile: detect
        # deletions and field changes instead of being add-only.
        self._marker_specs = {}

        if self._console is not None:
            # use a previously allocated console port
            if console_type == "vnc":
                vnc_console_start_port_range, vnc_console_end_port_range = self._get_vnc_console_port_range()
                self._console = self._manager.port_manager.reserve_tcp_port(
                    self._console,
                    self._project,
                    port_range_start=vnc_console_start_port_range,
                    port_range_end=vnc_console_end_port_range,
                )
            elif console_type == "none":
                self._console = None
            else:
                self._console = self._manager.port_manager.reserve_tcp_port(self._console, self._project)

        if self._aux is not None:
            # use a previously allocated auxiliary console port
            if aux_type == "vnc":
                # VNC is a special case and the range must be 5900-6000
                self._aux = self._manager.port_manager.reserve_tcp_port(
                    self._aux, self._project, port_range_start=5900, port_range_end=6000
                )
            elif aux_type == "none":
                self._aux = None
            else:
                self._aux = self._manager.port_manager.reserve_tcp_port(self._aux, self._project)

        if self._console is None:
            # allocate a new console
            if console_type == "vnc":
                vnc_console_start_port_range, vnc_console_end_port_range = self._get_vnc_console_port_range()
                self._console = self._manager.port_manager.get_free_tcp_port(
                    self._project,
                    port_range_start=vnc_console_start_port_range,
                    port_range_end=vnc_console_end_port_range,
                )
            elif console_type != "none":
                self._console = self._manager.port_manager.get_free_tcp_port(self._project)

        if self._aux is None:
            # allocate a new auxiliary console
            if aux_type == "vnc":
                # VNC is a special case and the range must be 5900-6000
                self._aux = self._manager.port_manager.get_free_tcp_port(
                    self._project, port_range_start=5900, port_range_end=6000
                )
            elif aux_type != "none":
                self._aux = self._manager.port_manager.get_free_tcp_port(self._project)

        if self._wrap_console:
            self._internal_console_port = self._manager.port_manager.get_free_tcp_port(self._project)

        if self._wrap_aux:
            self._internal_aux_port = self._manager.port_manager.get_free_tcp_port(self._project)

        log.debug(
            "{module}: {name} [{id}] initialized. Console port {console}".format(
                module=self.manager.module_name, name=self.name, id=self.id, console=self._console
            )
        )

    def __del__(self):

        if hasattr(self, "_temporary_directory") and self._temporary_directory is not None:
            if os.path.exists(self._temporary_directory):
                shutil.rmtree(self._temporary_directory, ignore_errors=True)

    @property
    def linked_clone(self):
        return self._linked_clone

    @linked_clone.setter
    def linked_clone(self, val):
        self._linked_clone = val

    @property
    def custom_adapters(self):
        return self._custom_adapters

    @custom_adapters.setter
    def custom_adapters(self, val):
        self._custom_adapters = val

    @property
    def status(self):
        """
        Returns current node status
        """

        return self._node_status

    @status.setter
    def status(self, status):

        self._node_status = status
        self.updated()

    def updated(self):
        """
        Sends an updated event
        """
        self.project.emit("node.updated", self)

    @property
    def command_line(self):
        """
        Returns command used to start the node
        """

        return self._command_line

    @command_line.setter
    def command_line(self, command_line):

        self._command_line = command_line

    @property
    def project(self):
        """
        Returns the node current project.

        :returns: Project instance.
        """

        return self._project

    @property
    def name(self):
        """
        Returns the name for this node.

        :returns: name
        """

        return self._name

    @name.setter
    def name(self, new_name):
        """
        Sets the name of this node.

        :param new_name: name
        """

        log.info(
            "{module}: {name} [{id}] renamed to {new_name}".format(
                module=self.manager.module_name, name=self.name, id=self.id, new_name=new_name
            )
        )
        self._name = new_name

    @property
    def usage(self):
        """
        Returns the usage for this node.

        :returns: usage
        """

        return self._usage

    @usage.setter
    def usage(self, new_usage):
        """
        Sets the usage of this node.

        :param new_usage: usage
        """

        self._usage = new_usage

    @property
    def id(self):
        """
        Returns the ID for this node.

        :returns: Node identifier (string)
        """

        return self._id

    @property
    def manager(self):
        """
        Returns the manager for this node.

        :returns: instance of manager
        """

        return self._manager

    @property
    def working_dir(self):
        """
        Return the node working directory
        """

        return self._project.node_working_directory(self)

    @property
    def working_path(self):
        """
        Return the node working path. Doesn't create structure of directories when not present.
        """

        return self._project.node_working_path(self)

    @property
    def temporary_directory(self):
        if self._temporary_directory is None:
            try:
                self._temporary_directory = tempfile.mkdtemp()
            except OSError as e:
                raise NodeError(f"Can't create temporary directory: {e}")
        return self._temporary_directory

    def create(self):
        """
        Creates the node.
        """

        log.debug("{module}: {name} [{id}] created".format(module=self.manager.module_name, name=self.name, id=self.id))

    async def delete(self):
        """
        Delete the node (including all its files).
        """

        def set_rw(operation, name, exc):
            # Add the missing user permissions instead of replacing the whole mode: chmod'ing
            # a directory to S_IWRITE removes the search permission on POSIX systems, making
            # the directory impossible to traverse or to delete. Also note that S_IWRITE
            # clears the read-only attribute on Windows.
            try:
                if os.path.isdir(name) and not os.path.islink(name):
                    os.chmod(name, os.stat(name).st_mode | stat.S_IRWXU)
                elif not os.path.islink(name):
                    os.chmod(name, os.stat(name).st_mode | stat.S_IRUSR | stat.S_IWUSR)
                # retry the failed operation now that the permissions are fixed
                # (retrying os.scandir is not possible, the directory is handled by the
                # retry loop below)
                if operation in (os.unlink, os.rmdir):
                    operation(name)
            except OSError:
                pass

        directory = self.project.node_working_directory(self)
        # Retry the deletion: a concurrent task (e.g. a MD5 checksum computation caching
        # its result in the node directory) can recreate a file while the directory is
        # being deleted, and shutil.rmtree silently gives up when its error handler returns
        for attempt in range(3):
            if not os.path.exists(directory):
                return
            try:
                await wait_run_in_executor(shutil.rmtree, directory, onerror=set_rw)
            except OSError as e:
                raise ComputeError(f"Could not delete the node working directory: {e}")
            if not os.path.exists(directory):
                return
            if attempt == 2:
                raise ComputeError(
                    f"Could not delete the node working directory '{directory}': a file may have been "
                    "recreated in it or could not be removed"
                )
            log.warning(f"Could not completely delete the node working directory '{directory}', retrying")
            await asyncio.sleep(0.1)

    def start(self):
        """
        Starts the node process.
        """

        raise NotImplementedError

    async def stop(self):
        """
        Stop the node process.
        """

        try:
            await self.stop_wrap_console()
        finally:
            self.status = "stopped"

    def suspend(self):
        """
        Suspends the node process.
        """

        raise NotImplementedError

    async def close(self):
        """
        Close the node process.
        """

        if self._closed:
            return False

        log.debug(
            "{module}: '{name}' [{id}]: is closing".format(module=self.manager.module_name, name=self.name, id=self.id)
        )

        if self._console:
            self._manager.port_manager.release_tcp_port(self._console, self._project)
            self._console = None
        if self._wrap_console:
            self._manager.port_manager.release_tcp_port(self._internal_console_port, self._project)
            self._internal_console_port = None
        if self._aux:
            self._manager.port_manager.release_tcp_port(self._aux, self._project)
            self._aux = None
        if self._wrap_aux:
            self._manager.port_manager.release_tcp_port(self._internal_aux_port, self._project)
            self._internal_aux_port = None

        self._closed = True
        return True

    def _get_vnc_console_port_range(self):
        """
        Returns the VNC console port range.
        """

        vnc_console_start_port_range = self._manager.config.settings.Server.vnc_console_start_port_range
        vnc_console_end_port_range = self._manager.config.settings.Server.vnc_console_end_port_range

        if not 5900 <= vnc_console_start_port_range <= 65535:
            raise NodeError("The VNC console start port range must be between 5900 and 65535")
        if not 5900 <= vnc_console_end_port_range <= 65535:
            raise NodeError("The VNC console start port range must be between 5900 and 65535")
        if vnc_console_start_port_range >= vnc_console_end_port_range:
            raise NodeError(
                f"The VNC console start port range value ({vnc_console_start_port_range}) "
                f"cannot be above or equal to the end value ({vnc_console_end_port_range})"
            )

        return vnc_console_start_port_range, vnc_console_end_port_range

    async def _wrap_console_proxy(self, internal_port, external_port, console_type):
        """
        Start a console proxy allowing multiple external clients to be
        connected at the same time.
        """

        remaining_trial = 60
        while True:
            try:
                (self._wrap_console_reader, self._wrap_console_writer) = await asyncio.open_connection(
                    host="127.0.0.1",
                    port=internal_port
                )
                break
            except (OSError, ConnectionRefusedError) as e:
                if remaining_trial <= 0:
                    raise e
            await asyncio.sleep(0.1)
            remaining_trial -= 1
        if console_type == "telnet":
            await AsyncioTelnetServer.write_client_intro(self._wrap_console_writer, echo=True)
            server = AsyncioTelnetServer(
                reader=self._wrap_console_reader,
                writer=self._wrap_console_writer,
                binary=True,
                echo=True
            )
        elif console_type == "ssh":
            server = AsyncioSSHServer(reader=self._wrap_console_reader, writer=self._wrap_console_writer)
        else:
            raise NodeError(f"Console wrapper does not support type {console_type}")

        # warning: this will raise OSError exception if there is a problem...
        proxy_server = await server.start(self._manager.port_manager.console_host, external_port)
        self._wrapper_telnet_servers.append(proxy_server)

    async def start_wrap_console(self):
        """
        Start console proxy servers for the console and auxiliary console allowing multiple clients
        to be connected at the same time
        """

        if self._wrap_console and self._console_type in ("telnet", "ssh"):
            await self._wrap_console_proxy(self._internal_console_port, self.console, self._console_type)
            log.info(
                f"New {self._console_type.upper()} proxy server for console started "
                f"(internal port = {self._internal_console_port}, external port = {self.console})"
            )

        if self._wrap_aux and self._aux_type in ("telnet", "ssh"):
            await self._wrap_console_proxy(self._internal_aux_port, self.aux, self._aux_type)
            log.info(
                f"New {self._aux_type.upper()} proxy server for auxiliary console started "
                f"(internal port = {self._internal_aux_port}, external port = {self.aux})"
            )

    async def stop_wrap_console(self):
        """
        Stops the console proxy servers.
        """

        if self._wrap_console_writer:
            self._wrap_console_writer.close()
            try:
                await self._wrap_console_writer.wait_closed()
            except (ConnectionResetError, BrokenPipeError, OSError):
                # Connection already closed, ignore error
                pass
        for telnet_proxy_server in self._wrapper_telnet_servers:
            telnet_proxy_server.close()
            try:
                await telnet_proxy_server.wait_closed()
            except (ConnectionResetError, BrokenPipeError, OSError):
                # Connection already closed, ignore error
                pass
        self._wrapper_telnet_servers = []

    async def reset_wrap_console(self):
        """
        Reset the wrap console (restarts the console proxy)
        """

        await self.stop_wrap_console()
        await self.start_wrap_console()

    async def start_websocket_console(self, websocket):
        """
        Connect to console using Websocket.

        :param websocket: FastAPI WebSocket object
        """

        log.info(
            f"New client {websocket.client.host}:{websocket.client.port}  has connected to compute"
            f" console WebSocket"
        )

        if self.status != "started":
            await websocket.close(code=1000)
            log.warning(f"Cannot open console WebSocket: node {self.name} is not started")
            return

        if self._console_type not in ("telnet", "ssh", "docker_exec"):
            await websocket.close(code=1000)
            log.warning(
                f"Cannot open console WebSocket: node {self.name} console type '{self._console_type}' "
                f"is not supported"
            )
            return

        telnet_reader = None
        telnet_writer = None
        ssh_connection = None
        ssh_process = None

        try:
            host = self._manager.port_manager.console_host
            if self._console_type == "ssh":
                # For SSH consoles (wrapped or not), connect to the external SSH port via an
                # SSH client. The AsyncioSSHServer on that port handles multi-client broadcasting
                # to/from the node's process streams. Connecting directly to _internal_console_port
                # would conflict with the exclusive connection already held by the SSH proxy.
                port = self.console
                ssh_connection = await asyncssh.connect(
                    host,
                    port=port,
                    username="gns3",
                    known_hosts=None,
                    encoding=None,
                )
                ssh_process = await ssh_connection.create_process(encoding=None, term_type="xterm")
                telnet_reader = ssh_process.stdout
                telnet_writer = ssh_process.stdin
            else:
                port = self.console
                (telnet_reader, telnet_writer) = await asyncio.open_connection(host, port)

            log.info(f"Connected to local console stream {host}:{port} (console type={self._console_type})")
        except (ConnectionError, OSError, asyncssh.Error) as e:
            await websocket.close(code=1000)
            log.warning(f"Cannot connect to node {self.name} console server: {e}")
            return

        def _parse_terminal_size_message(data: bytes):
            """
            Binary control frames sent by WebSocket console clients to propagate
            their terminal geometry: {"cols": int, "rows": int}. Terminal data
            travels as text frames (xterm.js AttachAddon), so binary frames are
            an unambiguous side channel. Returns (cols, rows) or None.
            """

            try:
                message = json.loads(data.decode("utf-8"))
            except (UnicodeDecodeError, ValueError):
                return None
            if not isinstance(message, dict):
                return None
            cols, rows = message.get("cols"), message.get("rows")
            if (
                isinstance(cols, int) and not isinstance(cols, bool)
                and isinstance(rows, int) and not isinstance(rows, bool)
                and 2 <= cols <= 5000
                and 2 <= rows <= 100000
            ):
                return cols, rows
            return None

        async def resize_console(cols: int, rows: int) -> None:
            """
            Propagate a client terminal resize to the node console stream:
            SSH channels use a pty request update, telnet-based consoles
            (including docker_exec) speak a NAWS subnegotiation to the console
            telnet server, which resizes the underlying stream (e.g. the
            docker exec pty).
            """

            if self._console_type == "ssh":
                with contextlib.suppress(AttributeError):
                    ssh_process.change_terminal_size(cols, rows)
            else:
                telnet_writer.write(
                    bytes([255, 251, 31])  # IAC WILL NAWS
                    + bytes([255, 250, 31])  # IAC SB NAWS
                    + struct.pack("!HH", cols, rows).replace(b"\xff", b"\xff\xff")
                    + bytes([255, 240])  # IAC SE
                )
                await telnet_writer.drain()

        async def ws_forward(telnet_writer):

            try:
                while True:
                    msg = await websocket.receive()
                    if msg["type"] == "websocket.disconnect":
                        break
                    if "text" in msg and msg["text"]:
                        data = msg["text"].encode()
                    elif "bytes" in msg and msg["bytes"]:
                        size = _parse_terminal_size_message(msg["bytes"])
                        if size is not None:
                            log.debug(
                                f"Console WebSocket client {websocket.client.host}:{websocket.client.port}"
                                f" resized terminal to {size[0]}x{size[1]}"
                            )
                            await resize_console(*size)
                            continue
                        data = msg["bytes"]
                    else:
                        continue
                    telnet_writer.write(data)
                    await telnet_writer.drain()
            except WebSocketDisconnect:
                log.info(
                    f"Client {websocket.client.host}:{websocket.client.port} has disconnected from compute"
                    f" console WebSocket"
                )

        async def telnet_forward(telnet_reader):

            try:
                while not telnet_reader.at_eof():
                    data = await telnet_reader.read(1024)
                    if data:
                        await websocket.send_bytes(data)
            except WebSocketDisconnect:
                # the client disconnected while node output was still streaming:
                # normal end of the session, not an error. Starlette raises
                # WebSocketDisconnect (whose str() is empty) from send once the
                # peer is gone, which used to surface as a message-less warning.
                log.info(
                    f"Client {websocket.client.host}:{websocket.client.port} has disconnected from compute"
                    f" console WebSocket while node output was being forwarded"
                )

        # keep forwarding websocket data in both direction
        if sys.version_info >= (3, 11, 0):
            # Starting with Python 3.11, passing coroutine objects to wait() directly is forbidden.
            aws = [asyncio.create_task(ws_forward(telnet_writer)), asyncio.create_task(telnet_forward(telnet_reader))]
        else:
            aws = [ws_forward(telnet_writer), telnet_forward(telnet_reader)]

        done, pending = await asyncio.wait(aws, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            if task.exception():
                log.warning(
                    f"Exception while forwarding WebSocket data to "
                    f"{self._console_type.upper()} server: {task.exception()!r}"
                )
        for task in pending:
            task.cancel()

        if ssh_connection:
            if ssh_process:
                ssh_process.close()
            ssh_connection.close()
            await ssh_connection.wait_closed()

        if telnet_writer and hasattr(telnet_writer, "close"):
            telnet_writer.close()
            if hasattr(telnet_writer, "wait_closed"):
                await telnet_writer.wait_closed()

    async def start_vnc_websocket_console(self, websocket):
        """
        Connect to VNC console using WebSocket.

        :param websocket: FastAPI WebSocket object
        """

        log.info(
            f"New client {websocket.client.host}:{websocket.client.port} has connected to compute "
            f"VNC console WebSocket"
        )

        if self.status != "started":
            await websocket.close(code=1000)
            log.warning(f"Cannot open VNC WebSocket: node {self.name} is not started")
            return
        if self._console_type != "vnc":
            await websocket.close(code=1000)
            log.warning(
                f"Cannot open VNC WebSocket: node {self.name} console type '{self._console_type}' is not vnc"
            )
            return

        try:
            vnc_reader, vnc_writer = await asyncio.open_connection(
                self._manager.port_manager.console_host,
                self.console  # VNC port
            )
            log.info(f"Connected to VNC server {self._manager.port_manager.console_host}:{self.console}")
        except ConnectionError as e:
            await websocket.close(code=1000)
            log.warning(f"Cannot connect to node {self.name} VNC server: {e}")
            return

        async def ws_forward(vnc_writer):
            # Browser → VNC: Forward binary WebSocket data to VNC server
            try:
                while True:
                    data = await websocket.receive_bytes()
                    if data:
                        vnc_writer.write(data)
                        await vnc_writer.drain()
            except WebSocketDisconnect:
                log.info(
                    f"Client {websocket.client.host}:{websocket.client.port} has disconnected from compute "
                    f"VNC console WebSocket"
                )

        async def vnc_forward(vnc_reader):
            # VNC → Browser: Forward VNC frames to WebSocket
            try:
                while not vnc_reader.at_eof():
                    data = await vnc_reader.read(65536)  # Larger buffer for VNC frames
                    if data:
                        await websocket.send_bytes(data)
            except WebSocketDisconnect:
                # the browser disconnected while VNC frames were still streaming
                # (starlette raises WebSocketDisconnect with an empty str() from
                # send once the peer is gone — not an error)
                log.info(
                    f"Client {websocket.client.host}:{websocket.client.port} has disconnected from compute "
                    f"VNC console WebSocket while frames were being forwarded"
                )
            except Exception as e:
                log.warning(f"Exception while forwarding VNC data to WebSocket: {e!r}")

        # Keep forwarding WebSocket data in both directions
        if sys.version_info >= (3, 11, 0):
            # Starting with Python 3.11, passing coroutine objects to wait() directly is forbidden.
            aws = [asyncio.create_task(ws_forward(vnc_writer)), asyncio.create_task(vnc_forward(vnc_reader))]
        else:
            aws = [ws_forward(vnc_writer), vnc_forward(vnc_reader)]

        done, pending = await asyncio.wait(aws, return_when=asyncio.FIRST_COMPLETED)
        for task in done:
            if task.exception():
                log.warning(f"Exception while forwarding WebSocket data to VNC server: {task.exception()!r}")
        for task in pending:
            task.cancel()

        vnc_writer.close()
        await vnc_writer.wait_closed()

    @property
    def aux(self):
        """
        Returns the aux console port of this node.

        :returns: aux console port
        """

        return self._aux

    @aux.setter
    def aux(self, aux):
        """
        Changes the aux port

        :params aux: Console port (integer) or None to free the port
        """

        if aux == self._aux or self._aux_type == "none":
            return

        if self._aux_type == "vnc" and aux is not None and aux < 5900:
            raise NodeError(f"VNC auxiliary console require a port superior or equal to 5900, current port is {aux}")

        if self._aux:
            self._manager.port_manager.release_tcp_port(self._aux, self._project)
            self._aux = None
        if aux is not None:
            if self.aux_type == "vnc":
                self._aux = self._manager.port_manager.reserve_tcp_port(
                    aux, self._project, port_range_start=5900, port_range_end=6000
                )
            else:
                self._aux = self._manager.port_manager.reserve_tcp_port(aux, self._project)

            log.info(
                "{module}: '{name}' [{id}]: auxiliary console port set to {port}".format(
                    module=self.manager.module_name, name=self.name, id=self.id, port=aux
                )
            )

    @property
    def console(self):
        """
        Returns the console port of this node.

        :returns: console port
        """

        return self._console

    @console.setter
    def console(self, console):
        """
        Changes the console port

        :params console: Console port (integer) or None to free the port
        """

        if console == self._console or self._console_type == "none":
            return

        if self._console_type == "vnc" and console is not None and console < 5900:
            raise NodeError(f"VNC console require a port superior or equal to 5900, current port is {console}")

        if self._console:
            self._manager.port_manager.release_tcp_port(self._console, self._project)
            self._console = None
        if console is not None:
            if self.console_type == "vnc":
                vnc_console_start_port_range, vnc_console_end_port_range = self._get_vnc_console_port_range()
                self._console = self._manager.port_manager.reserve_tcp_port(
                    console,
                    self._project,
                    port_range_start=vnc_console_start_port_range,
                    port_range_end=vnc_console_end_port_range,
                )
            else:
                self._console = self._manager.port_manager.reserve_tcp_port(console, self._project)

            log.info(
                "{module}: '{name}' [{id}]: console port set to {port}".format(
                    module=self.manager.module_name, name=self.name, id=self.id, port=console
                )
            )

    @property
    def console_type(self):
        """
        Returns the console type for this node.

        :returns: console type (string)
        """

        return self._console_type

    @console_type.setter
    def console_type(self, console_type):
        """
        Sets the console type for this node.

        :param console_type: console type (string)
        """

        if console_type != self._console_type:
            # get a new port if the console type change
            if self._console:
                self._manager.port_manager.release_tcp_port(self._console, self._project)
            if console_type == "none":
                # no need to allocate a port when the console type is none
                self._console = None
            elif console_type == "vnc":
                vnc_console_start_port_range, vnc_console_end_port_range = self._get_vnc_console_port_range()
                self._console = self._manager.port_manager.get_free_tcp_port(
                    self._project,
                    vnc_console_start_port_range,
                    vnc_console_end_port_range
                )
            else:
                self._console = self._manager.port_manager.get_free_tcp_port(self._project)

        self._console_type = console_type
        log.info(
            "{module}: '{name}' [{id}]: console type set to {console_type} (console port is {console})".format(
                module=self.manager.module_name,
                name=self.name,
                id=self.id,
                console_type=console_type,
                console=self.console,
            )
        )

    @property
    def aux_type(self):
        """
        Returns the auxiliary console type for this node.

        :returns: aux type (string)
        """

        return self._aux_type

    @aux_type.setter
    def aux_type(self, aux_type):
        """
        Sets the auxiliary console type for this node.

        :param aux_type: console type (string)
        """

        if aux_type != self._aux_type:
            # get a new port if the aux type change
            if self._aux:
                self._manager.port_manager.release_tcp_port(self._aux, self._project)
            if aux_type == "none":
                # no need to allocate a port when the auxiliary console type is none
                self._aux = None
            elif aux_type == "vnc":
                # VNC is a special case and the range must be 5900-6000
                self._aux = self._manager.port_manager.get_free_tcp_port(self._project, 5900, 6000)
            else:
                self._aux = self._manager.port_manager.get_free_tcp_port(self._project)

        self._aux_type = aux_type
        log.info(
            "{module}: '{name}' [{id}]: console type set to {aux_type} (auxiliary console port is {aux})".format(
                module=self.manager.module_name, name=self.name, id=self.id, aux_type=aux_type, aux=self.aux
            )
        )

    @property
    def ubridge(self):
        """
        Returns the uBridge hypervisor.

        :returns: instance of uBridge
        """

        if self._ubridge_hypervisor and not self._ubridge_hypervisor.is_running():
            self._ubridge_hypervisor = None
        return self._ubridge_hypervisor

    @ubridge.setter
    def ubridge(self, ubride_hypervisor):
        """
        Set an uBridge hypervisor.

        :param ubride_hypervisor: uBridge hypervisor
        """

        self._ubridge_hypervisor = ubride_hypervisor

    @property
    def ubridge_path(self):
        """
        Returns the uBridge executable path.

        :returns: path to uBridge
        """

        path = shutil.which(self._manager.config.settings.Server.ubridge_path)
        return path

    @locking
    async def _ubridge_send(self, command):
        """
        Sends a command to uBridge hypervisor.

        :param command: command to send
        """

        if not self._ubridge_hypervisor or not self._ubridge_hypervisor.is_running():
            await self._start_ubridge(self._ubridge_require_privileged_access)
        if not self._ubridge_hypervisor or not self._ubridge_hypervisor.is_running():
            raise NodeError(f"Cannot send command '{command}': uBridge is not running")
        try:
            await self._ubridge_hypervisor.send(command)
        except UbridgeError as e:
            raise UbridgeError(
                f"Error while sending command '{command}': {e}: {self._ubridge_hypervisor.read_stdout()}"
            )

    @locking
    async def _start_ubridge(self, require_privileged_access=False):
        """
        Starts uBridge (handles connections to and from this node).
        """

        # Prevent us to start multiple ubridge
        if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running():
            return

        if self.ubridge_path is None:
            raise NodeError(
                "uBridge is not available, path doesn't exist, or you just installed GNS3 and need to restart your user session to refresh user permissions."
            )

        if require_privileged_access and not self._manager.has_privileged_access(self.ubridge_path):
            raise NodeError("uBridge requires root access or the capability to interact with network adapters")

        server_host = self._manager.config.settings.Server.host
        transport = self._manager.config.settings.Server.ubridge_control_transport
        if not self.ubridge:
            self._ubridge_hypervisor = Hypervisor(
                self._project, self.ubridge_path, self.working_dir, transport, server_host, self.id
            )
        log.debug(f"Starting new uBridge hypervisor at {self._ubridge_hypervisor.endpoint}")
        await self._ubridge_hypervisor.start()
        if self._ubridge_hypervisor:
            log.info(
                f"Hypervisor at {self._ubridge_hypervisor.endpoint} has successfully started"
            )
            await self._ubridge_hypervisor.connect()
            # Tell this uBridge where to send MARK signals and which node id to
            # tag them with. Marker is opt-in and inert until a `mark` filter is
            # added, so this never disturbs the data plane.
            await self._ubridge_configure_marker_sink()
        # save if privileged are required in case uBridge needs to be restarted in self._ubridge_send()
        self._ubridge_require_privileged_access = require_privileged_access

    async def _ubridge_configure_marker_sink(self):
        """
        Point this node's uBridge at the compute's marker UDP sink and tag its
        signals with this node's id. Safe to call before any marker filter
        exists — uBridge stays inert until a ``mark`` filter is configured.

        Old uBridge builds without the marker module are tolerated: the failure
        is downgraded to a warning so node start is not blocked by an opt-in
        observability feature.
        """

        from gns3server.compute.marker.marker_manager import MarkerManager

        manager = MarkerManager.instance()
        if not manager.running or not manager.host or not manager.port:
            return
        if self._ubridge_hypervisor is None:
            return
        try:
            # Talk to the hypervisor directly, NOT via _ubridge_send: this runs
            # inside _start_ubridge, which is reached THROUGH _ubridge_send when
            # uBridge starts lazily (e.g. linking a stopped node). _ubridge_send's
            # lock is non-reentrant, so calling it again here would deadlock on
            # the held ___ubridge_send_lock. uBridge is already running and
            # connected at this point, so the raw hypervisor send is safe.
            await self._ubridge_hypervisor.send(f"marker sink {manager.host} {manager.port}")
            await self._ubridge_hypervisor.send(f"marker node {self._id}")
        except UbridgeError:
            log.warning(
                "uBridge does not support the marker module; traffic insight disabled for node %r",
                self.name,
            )

    async def _stop_ubridge(self):
        """
        Stops uBridge.
        """

        if self._ubridge_hypervisor and self._ubridge_hypervisor.is_running():
            log.debug(f"Stopping uBridge hypervisor at {self._ubridge_hypervisor.endpoint}")
            await self._ubridge_hypervisor.stop()
        self._ubridge_hypervisor = None
        # uBridge is gone, so every marker filter (and its in-bridge state) is
        # gone too — clear the map so the next apply re-installs them all rather
        # than skipping them as "already installed".
        self._marker_filter_bridges.clear()
        self._marker_specs.clear()

    async def add_ubridge_udp_connection(self, bridge_name, source_nio, destination_nio):
        """
        Creates an UDP connection in uBridge.

        :param bridge_name: bridge name in uBridge
        :param source_nio: source NIO instance
        :param destination_nio: destination NIO instance
        """

        await self._ubridge_send(f"bridge create {bridge_name}")

        if not isinstance(destination_nio, NIOUDP):
            raise NodeError("Destination NIO is not UDP")

        await self._ubridge_send(
            "bridge add_nio_udp {name} {lport} {rhost} {rport}".format(
                name=bridge_name, lport=source_nio.lport, rhost=source_nio.rhost, rport=source_nio.rport
            )
        )

        await self._ubridge_send(
            "bridge add_nio_udp {name} {lport} {rhost} {rport}".format(
                name=bridge_name, lport=destination_nio.lport, rhost=destination_nio.rhost, rport=destination_nio.rport
            )
        )

        if destination_nio.capturing:
            await self._ubridge_send(
                'bridge start_capture {name} "{pcap_file}"'.format(
                    name=bridge_name, pcap_file=destination_nio.pcap_output_file
                )
            )

        await self._ubridge_send(f"bridge start {bridge_name}")
        await self._ubridge_apply_filters(bridge_name, destination_nio.filters)
        await self._ubridge_apply_markers(bridge_name, destination_nio)

    async def update_ubridge_udp_connection(self, bridge_name, source_nio, destination_nio):
        if destination_nio:
            await self._ubridge_apply_filters(bridge_name, destination_nio.filters)
            await self._ubridge_apply_markers(bridge_name, destination_nio)

    async def ubridge_delete_bridge(self, name):
        """
        :params name: Delete the bridge with this name
        """

        if self.ubridge:
            await self._ubridge_send(f"bridge delete {name}")

    async def _ubridge_apply_filters(self, bridge_name, filters):
        """
        Apply packet filters

        :param bridge_name: bridge name in uBridge
        :param filters: Array of filter dictionary
        """

        await self._ubridge_send("bridge reset_packet_filters " + bridge_name)
        for packet_filter in self._build_filter_list(filters):
            cmd = f"bridge add_packet_filter {bridge_name} {packet_filter}"
            try:
                await self._ubridge_send(cmd)
            except UbridgeError as e:
                match = re.search(r"Cannot compile filter '(.*)': syntax error", str(e))
                if match:
                    message = f"Warning: ignoring BPF packet filter '{self.name}' due to syntax error: {match.group(1)}"
                    log.warning(message)
                    self.project.emit("log.warning", {"message": message})
                else:
                    raise

    def _build_filter_list(self, filters):
        """
        :returns: Iterator building a list of filter
        """

        i = 0
        for (filter_type, values) in filters.items():
            if isinstance(values[0], str):
                for line in values[0].split("\n"):
                    line = line.strip()
                    yield "{filter_name} {filter_type} {filter_value}".format(
                        filter_name="filter" + str(i),
                        filter_type=filter_type,
                        filter_value='"{}" {}'.format(line, " ".join([str(v) for v in values[1:]])),
                    ).strip()
                    i += 1
            else:
                yield "{filter_name} {filter_type} {filter_value}".format(
                    filter_name="filter" + str(i),
                    filter_type=filter_type,
                    filter_value=" ".join([str(v) for v in values]),
                )
                i += 1

    @staticmethod
    def _marker_linktype(data_link_type):
        """
        Normalize a GNS3 pcap data-link type (e.g. ``DLT_C_HDLC``) to the bare
        uBridge ``linktype`` token (``C_HDLC``) by stripping the ``DLT_`` prefix.
        Returns ``None`` for Ethernet (``DLT_EN10MB`` / unset) so the ``linktype``
        keyword is omitted and uBridge defaults to EN10MB. Values come straight
        from ``SerialPort.data_link_types`` (the single source of truth); uBridge
        resolves them with ``pcap_datalink_name_to_val``, which is case-sensitive
        and expects the canonical uppercase form.
        """
        if not data_link_type:
            return None
        dlt = data_link_type.upper()
        if dlt.startswith("DLT_"):
            dlt = dlt[4:]
        return None if dlt == "EN10MB" else dlt

    async def _ubridge_add_marker_filter(self, bridge_name, name, bpf, pcap_path, tag=None, link_id=None, direction=None, data_link_type=None):
        """
        Attach a `mark` packet filter to a uBridge bridge for traffic insight.

        On BPF match uBridge (a) emits a UDP MARK signal to the configured sink
        and (b) appends the packet to ``pcap_path``. Unlike the impairment
        filters, this is an observability tap: it never drops or alters traffic,
        and it is added/removed on its own (not via reset_packet_filters) so the
        pcap is not closed/reopened on unrelated filter changes.

        :param bridge_name: uBridge bridge carrying the link's traffic
        :param name: stable, gns3server-chosen filter name (pcap identity + echoed in signals)
        :param bpf: libpcap BPF expression
        :param pcap_path: absolute path ubridge appends matched packets to
        :param tag: optional correlation id echoed in MARK signals
        """

        # mark <bpf> [tag <id>] [pcap <path>] — tag/pcap keyword pairs, any order.
        # name travels from the controller REST layer (MarkerCreate schema) but is
        # validated here too as defense-in-depth against hand-edited topology files.
        # Note: "global-*" names are legitimate here — they come from project-level
        # marker definitions (inherit_marker). The prefix is only forbidden at the
        # user-facing schema layer, not at the uBridge boundary.
        _MARKER_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
        # Defense-in-depth vs hand-edited topology: the user-facing name is capped
        # at 32 by the schema; inherited copies carry a ``global-`` prefix (≤ 39),
        # so allow up to 48 here.
        if not _MARKER_NAME_RE.match(name) or len(name) > 48:
            raise UbridgeError(f"Invalid marker name: {name!r}")
        cmd = 'bridge add_packet_filter {bridge} {name} mark "{bpf}"'.format(
            bridge=bridge_name, name=name, bpf=bpf
        )
        if tag is not None:
            cmd += f" tag {tag}"
        # Per-link attribution (contract §3.2): when one ubridge bridge serves
        # several GNS3 links (e.g. IOU's per-node bridge), bridge+filter collide,
        # so the link id is the only way to tell signals — and pcap files — apart.
        if link_id:
            cmd += f" link {link_id}"
        if direction is not None:
            cmd += f" dir {direction}"
        linktype = self._marker_linktype(data_link_type)
        if linktype is not None:
            cmd += f" linktype {linktype}"
        cmd += ' pcap "{path}"'.format(path=pcap_path)
        # Let BPF compile errors propagate — the marker is the user's intent, so a
        # bad expression must surface instead of being silently dropped.
        await self._ubridge_send(cmd)

    async def delete_marker_capture(self, name, link_id, nio=None):
        """
        Remove a marker from uBridge (fine-grained ``delete_packet_filter`` — NOT
        reset_packet_filters, so sibling markers' pcaps aren't closed/reopened)
        and delete its capture pcap. Called by the controller when a marker is
        removed; safe with the node stopped (filter removal is skipped, the file
        is still unlinked). IOU overrides ``_ubridge_delete_marker_filter`` for
        its ``iol_bridge`` command shape.

        ``nio`` is the port NIO whose cached ``nio.markers`` carries this marker
        spec; it is dropped here so a later node start / NIO reapply
        (``_ubridge_apply_markers``) does not reinstall the marker. Without this,
        deleting a marker while the node is stopped left the spec in
        ``nio.markers``, and starting the node recreated an empty pcap.
        """
        if nio is not None and getattr(nio, "markers", None):
            nio.markers.pop(name, None)
        bridge_name = self._marker_filter_bridges.pop((name, link_id), None)
        self._marker_specs.pop((name, link_id), None)
        if bridge_name is not None:
            await self._ubridge_delete_marker_filter(bridge_name, name)
        try:
            markers_dir = self.project.markers_working_directory()
            pcap_path = os.path.join(markers_dir, f"{self._id}_{link_id}_{name}.pcap")
            os.remove(pcap_path)
        except FileNotFoundError:
            pass
        except OSError as e:
            log.warning("Could not remove marker pcap for '%s' on link %s: %s", name, link_id, e)

    async def _ubridge_delete_marker_filter(self, bridge_name, name):
        """
        Remove a single marker filter from uBridge with ``delete_packet_filter``
        (not a bridge-wide reset) so other markers keep their pcaps open. A no-op
        when uBridge isn't running — the pcap cleanup in the caller still proceeds.
        """
        if not (self._ubridge_hypervisor and self._ubridge_hypervisor.is_running()):
            return
        try:
            await self._ubridge_send(f"bridge delete_packet_filter {bridge_name} {name}")
        except UbridgeError as e:
            log.warning("Could not remove marker filter '%s' from %s: %s", name, bridge_name, e)

    async def rebuild_marker_filter(self, name, link_id, bpf, tag=None, direction=None, enabled=True):
        """
        Re-install a single marker filter with new params (delete + add), without
        a bridge-wide reset — so sibling markers keep their pcaps open. uBridge
        reopens the marker's own pcap on re-add (a new capture session for the
        new BPF), which is expected. No-op if the marker isn't installed (node
        stopped) — the next NIO reapply picks up the updated ``_markers``.

        IOU needs no override: this calls ``_ubridge_delete_marker_filter`` /
        ``_ubridge_add_marker_filter`` / ``_ubridge_set_marker_filter_state``,
        all of which IOU already overrides for ``iol_bridge``.
        """
        bridge_name = self._marker_filter_bridges.get((name, link_id))
        if bridge_name is None:
            return
        await self._ubridge_delete_marker_filter(bridge_name, name)
        pcap_path = os.path.join(self.project.markers_working_directory(), f"{self._id}_{link_id}_{name}.pcap")
        await self._ubridge_add_marker_filter(bridge_name, name, bpf, pcap_path, tag, link_id, direction=direction)
        if not enabled:
            await self._ubridge_set_marker_filter_state(name, enabled=False)

    async def _ubridge_apply_markers(self, bridge_name, nio):
        """
        Reconcile the traffic-insight markers carried by *nio* onto bridge
        *bridge_name* with what is already installed there.

        uBridge's ``reset_packet_filters`` preserves mark filters (contract), so
        a plain re-add would duplicate them; instead this diffs the desired
        ``nio.markers`` against the installed ``_marker_specs``:

          * installed but no longer desired  → delete filter + unlink pcap
          * desired with changed bpf/tag/direction/data_link_type → rebuild
            (delete + add; the marker's own pcap reopens for the new BPF)
          * desired with only ``enabled`` changed → instant on/off toggle
            (sibling and own pcap stay open)
          * desired and unchanged              → skip
          * desired and new                    → add

        Called from ``add_ubridge_udp_connection`` (fresh bridge, empty maps →
        installs all) and ``update_ubridge_udp_connection`` / the batch NIO
        update path (incremental reconcile).
        """
        from gns3server.compute.marker.marker_manager import MarkerManager

        markers = nio.markers if hasattr(nio, 'markers') else {}
        manager = MarkerManager.instance()
        markers_dir = self.project.markers_working_directory()
        desired = {(name, spec.get("link_id", "")): spec for name, spec in markers.items()}

        # 1. Remove installed markers that are no longer desired (marker/def delete).
        # Scope to THIS bridge: the map is node-wide and also holds markers
        # installed on this node's other links/NIOs. Without this guard,
        # reconciling one NIO would delete every other link's markers + pcaps
        # (desired only carries the current NIO's markers) — a regression.
        for key in list(self._marker_filter_bridges):
            if self._marker_filter_bridges[key] != bridge_name:
                continue
            if key not in desired:
                mname, link_id = key
                installed_bridge = self._marker_filter_bridges.pop(key)
                self._marker_specs.pop(key, None)
                await self._ubridge_delete_marker_filter(installed_bridge, mname)
                try:
                    os.remove(os.path.join(markers_dir, f"{self._id}_{link_id}_{mname}.pcap"))
                except FileNotFoundError:
                    pass
                except OSError as e:
                    log.warning("Could not remove marker pcap for '%s' on link %s: %s", mname, link_id, e)
                manager.unregister(self._id, mname)

        # 2. Add newly-desired markers; rebuild ones whose filter fields changed.
        rebuild_fields = ("bpf", "tag", "direction", "data_link_type")
        for (name, link_id), spec in desired.items():
            bpf = spec.get("bpf", "")
            tag = spec.get("tag")
            enabled = spec.get("enabled", True)
            if (name, link_id) in self._marker_filter_bridges:
                installed_spec = self._marker_specs.get((name, link_id))
                if installed_spec is None:
                    # Installed but no recorded spec (legacy / pre-reconcile state):
                    # cannot diff, skip to avoid a duplicate add.
                    continue
                if any(installed_spec.get(f) != spec.get(f) for f in rebuild_fields):
                    # A filter field changed → rebuild (delete + re-add).
                    installed_bridge = self._marker_filter_bridges.get((name, link_id))
                    await self._ubridge_delete_marker_filter(installed_bridge, name)
                elif installed_spec.get("enabled", True) != enabled:
                    # Only the on/off state changed → instant toggle, pcap preserved.
                    await self._ubridge_set_marker_filter_state(name, enabled)
                    self._marker_specs[(name, link_id)] = spec
                    continue
                else:
                    continue  # unchanged
            pcap_path = os.path.join(markers_dir, f"{self._id}_{link_id}_{name}.pcap")
            try:
                await self._ubridge_add_marker_filter(bridge_name, name, bpf, pcap_path, tag, link_id,
                                                     direction=spec.get("direction"),
                                                     data_link_type=spec.get("data_link_type"))
            except UbridgeError as e:
                # Swallow BPF compile errors (warn + skip) so a single bad
                # expression can't break link creation / node restart — mirrors
                # _ubridge_apply_filters, which does the same for packet filters.
                if "syntax error" in str(e).lower() or "compile filter" in str(e).lower():
                    message = f"Warning: ignoring marker '{name}' due to BPF syntax error: {e}"
                    log.warning(message)
                    self.project.emit("log.warning", {"message": message})
                    continue
                raise
            # A disabled marker is installed but turned off (a paused tap), not
            # dropped — so the UI can flip it back on instantly with
            # enable_packet_filter, no NIO rebuild (ubridge contract §3.2).
            if not enabled:
                try:
                    await self._ubridge_send(f"bridge enable_packet_filter {bridge_name} {name} off")
                except UbridgeError as e:
                    # Old ubridge without enable_packet_filter: leave it installed
                    # (on) rather than fail the whole link/marker apply.
                    log.warning(f"Could not turn marker '{name}' off on {bridge_name}: {e}")
            manager.register(str(self.project.id), self._id, name, link_id, tag)
            # Remember which bridge hosts this filter so an instant on/off toggle
            # (no NIO rebuild) can resolve it by name alone, and keep the spec so
            # the next reconcile can detect changes.
            self._marker_filter_bridges[name, link_id] = bridge_name
            self._marker_specs[name, link_id] = spec

    async def _ubridge_set_marker_filter_state(self, name, enabled):
        """
        Toggle an installed marker filter on/off with a single uBridge command
        (``bridge enable_packet_filter … on|off``) — no NIO reset/reapply, so the
        pcap identity and emitted counter are preserved (ubridge contract §3.2).
        The bridge is resolved from the (name, link_id)→bridge map populated at
        apply time; entries are iterated so a node that hosts the same marker name
        on several links (e.g. IOU with one IOL-BRIDGE per node) toggles every
        copy. IOU overrides this for its ``iol_bridge`` command shape.

        :param name: marker filter name
        :param enabled: True = on (signal+pcap), False = off (paused tap)
        """

        state = "on" if enabled else "off"
        for (n, lid), bridge_name in list(self._marker_filter_bridges.items()):
            if n == name:
                await self._ubridge_send(f"bridge enable_packet_filter {bridge_name} {name} {state}")

    async def _ubridge_marker_pause(self):
        """
        Pause all marker signal+pcap emission on this node's uBridge
        (``marker pause``). Keeps the sink open so ``resume`` is instant. Safe
        on old ubridge builds (the error is downgraded to a warning). Called by
        the project-level pause fan-out.
        """

        if self._ubridge_hypervisor:
            try:
                await self._ubridge_hypervisor.send("marker pause")
            except UbridgeError as e:
                log.warning(f"Could not pause markers on node {self._id}: {e}")

    async def _ubridge_marker_resume(self):
        """Resume marker signal+pcap emission (``marker resume``)."""

        if self._ubridge_hypervisor:
            try:
                await self._ubridge_hypervisor.send("marker resume")
            except UbridgeError as e:
                log.warning(f"Could not resume markers on node {self._id}: {e}")

    async def _add_ubridge_ethernet_connection(self, bridge_name, ethernet_interface, block_host_traffic=False):
        """
        Creates a connection with an Ethernet interface in uBridge.

        :param bridge_name: bridge name in uBridge
        :param ethernet_interface: Ethernet interface name
        :param block_host_traffic: block network traffic originating from the host OS (Windows only)
        """

        if sys.platform.startswith("linux") and block_host_traffic is False:
            # on Linux we use RAW sockets by default excepting if host traffic must be blocked
            await self._ubridge_send(
                'bridge add_nio_linux_raw {name} "{interface}"'.format(name=bridge_name, interface=ethernet_interface)
            )
        else:
            # on other platforms we just rely on the pcap library
            await self._ubridge_send(
                'bridge add_nio_ethernet {name} "{interface}"'.format(name=bridge_name, interface=ethernet_interface)
            )
            source_mac = None
            for interface in interfaces():
                if interface["name"] == ethernet_interface:
                    source_mac = interface["mac_address"]
            if source_mac:
                await self._ubridge_send(
                    'bridge set_pcap_filter {name} "not ether src {mac}"'.format(name=bridge_name, mac=source_mac)
                )
                log.info(f"PCAP filter applied on '{ethernet_interface}' for source MAC {source_mac}")

    def _create_local_udp_tunnel(self):
        """
        Creates a local UDP tunnel (pair of 2 NIOs, one for each direction)

        :returns: source NIO and destination NIO.
        """

        m = PortManager.instance()
        lport = m.get_free_udp_port(self.project)
        rport = m.get_free_udp_port(self.project)
        source_nio_settings = {"lport": lport, "rhost": "127.0.0.1", "rport": rport, "type": "nio_udp"}
        destination_nio_settings = {"lport": rport, "rhost": "127.0.0.1", "rport": lport, "type": "nio_udp"}
        source_nio = self.manager.create_nio(source_nio_settings)
        destination_nio = self.manager.create_nio(destination_nio_settings)
        log.info(
            "{module}: '{name}' [{id}]:local UDP tunnel created between port {port1} and {port2}".format(
                module=self.manager.module_name, name=self.name, id=self.id, port1=lport, port2=rport
            )
        )
        return source_nio, destination_nio

    @property
    def hw_virtualization(self):
        """
        Returns either the node is using hardware virtualization or not.

        :return: boolean
        """

        return self._hw_virtualization

    def check_available_ram(self, requested_ram):
        """
        Sends a warning notification if there is not enough RAM on the system to allocate requested RAM.

        :param requested_ram: requested amount of RAM in MB
        """

        available_ram = int(psutil.virtual_memory().available / (1024 * 1024))
        percentage_left = 100 - psutil.virtual_memory().percent
        if requested_ram > available_ram:
            message = '"{}" requires {}MB of RAM to run but there is only {}MB - {}% of RAM left on "{}"'.format(
                self.name, requested_ram, available_ram, percentage_left, platform.node()
            )
            self.project.emit("log.warning", {"message": message})

    def _get_custom_adapter_settings(self, adapter_number):

        if self.custom_adapters:
            for custom_adapter in self.custom_adapters:
                if custom_adapter["adapter_number"] == adapter_number:
                    return custom_adapter
        return {}