gns3-server/gns3server/utils/interfaces.py

#
# Copyright (C) 2014 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 os
import socket
import struct
import sys

import psutil

from gns3server.compute.compute_error import ComputeError
from gns3server.config import Config

if psutil.version_info < (3, 0, 0):
    raise Exception(
        "psutil version should >= 3.0.0. If you are under Ubuntu/Debian install gns3 via apt instead of pip"
    )

import logging

log = logging.getLogger(__name__)


def has_netmask(interface_name):
    """
    Checks if an interface has a netmask.

    :param interface: interface name

    :returns: boolean
    """
    for interface in interfaces():
        if interface["name"] == interface_name:
            if interface["netmask"] and len(interface["netmask"]) > 0:
                return True
            return False
    return False


def is_interface_up(interface):
    """
    Checks if an interface is up.

    :param interface: interface name

    :returns: boolean
    """

    if sys.platform.startswith("linux") or sys.platform.startswith("openbsd"):
        if interface not in psutil.net_if_addrs():
            return False

        import fcntl

        SIOCGIFFLAGS = 0x8913
        try:
            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
                result = fcntl.ioctl(s.fileno(), SIOCGIFFLAGS, interface + "\0" * 256)
                (flags,) = struct.unpack("H", result[16:18])
                if flags & 1:  # check if the up bit is set
                    return True
            return False
        except OSError as e:
            raise ComputeError(f"Exception when checking if {interface} is up: {e}")
    else:
        # TODO: Windows & OSX support
        return True


def is_interface_bridge(interface):
    """
    :returns: True if interface is a bridge
    """
    return os.path.exists(os.path.join("/sys/class/net/", interface, "bridge"))


def interfaces():
    """
    Gets the network interfaces on this server.

    :returns: list of network interfaces
    """

    results = []
    allowed_interfaces = Config.instance().settings.Server.allowed_interfaces
    net_if_addrs = psutil.net_if_addrs()
    net_if_stats = psutil.net_if_stats()
    for interface in sorted(net_if_addrs.keys()):
        if allowed_interfaces and interface not in allowed_interfaces and not interface.startswith("gns3tap"):
            log.warning(f"Interface '{interface}' is not allowed to be used on this server")
            continue
        ip_address = ""
        mac_address = ""
        netmask = ""
        interface_type = "ethernet"
        # Collect every IPv4 and IPv6 address on this interface. An interface may
        # carry several addresses of each family (or none at all), so we keep a
        # list in addition to the legacy single IPv4 value retained for backward
        # compatibility with existing callers (compute link detection, GNS3 VM,
        # VMware, has_netmask(), ...).
        ip_addresses = []
        for addr in net_if_addrs[interface]:
            if addr.family == socket.AF_INET:
                # legacy single-value behavior (keeps the last IPv4 seen)
                ip_address = addr.address
                netmask = addr.netmask
                ip_addresses.append({"family": "ipv4", "address": addr.address, "netmask": addr.netmask or None})
            elif addr.family == socket.AF_INET6:
                ip_addresses.append({"family": "ipv6", "address": addr.address, "netmask": addr.netmask or None})
            if addr.family == psutil.AF_LINK:
                mac_address = addr.address
        if interface.startswith("tap"):
            # found no way to reliably detect a TAP interface
            interface_type = "tap"
        # Operational state and link attributes (speed/mtu/flags) come from
        # psutil.net_if_stats(). An interface present in net_if_addrs is normally
        # also present here; fall back to neutral defaults when it is not.
        status = "down"
        speed = 0
        mtu = 0
        flags = []
        stats = net_if_stats.get(interface)
        if stats is not None:
            status = "up" if stats.isup else "down"
            speed = stats.speed
            mtu = stats.mtu
            # psutil returns flags either as a comma-separated string (>= 6.0) or
            # as a list (older versions); normalize to a list for a stable shape.
            f = stats.flags
            if isinstance(f, str):
                flags = [flag for flag in f.split(",") if flag]
            else:
                flags = f
        results.append(
            {
                "id": interface,
                "name": interface,
                "ip_address": ip_address,
                "netmask": netmask,
                "mac_address": mac_address,
                "type": interface_type,
                "ip_addresses": ip_addresses,
                "status": status,
                "speed": speed,
                "mtu": mtu,
                "flags": flags,
            }
        )

    # This interface have special behavior
    for result in results:
        result["special"] = False
        for special_interface in (
            "lo",
            "vmnet",
            "vboxnet",
            "docker",
            "lxcbr",
            "virbr",
            "ovs-system",
            "veth",
            "fw",
            "p2p",
            "bridge",
            "vmware",
            "virtualbox",
            "gns3",
            "veb",
        ):
            if result["name"].lower().startswith(special_interface):
                result["special"] = True
        for special_interface in "-nic":
            if result["name"].lower().endswith(special_interface):
                result["special"] = True
    return results