<commit_before>"""
Detect physical devices that can be used by chutes.

This module detects physical devices (for now just network interfaces) that can
be used by chutes.  This includes WAN interfaces for Internet connectivity and
WiFi interfaces which can host APs.

It also makes sure certain entries exist in the system UCI files for these
devices, for example "wifi-device" sections.  These are shared between chutes,
so they only need to be added when missing.
"""

import itertools
import operator
import os
import re
import subprocess

from paradrop.base.output import out
from paradrop.base import settings
from paradrop.lib.utils import datastruct, pdos, uci
from paradrop.core.config import uciutils

IEEE80211_DIR = "/sys/class/ieee80211"
SYS_DIR = "/sys/class/net"
EXCLUDE_IFACES = set(["lo"])

# Strings that identify a virtual interface.
VIF_MARKERS = [".", "veth"]

# Matches various ways of specifying WiFi devices (phy0, wlan0).
WIFI_DEV_REF = re.compile("([a-z]+)(\d+)")

# Set of wifi-interface mode values that are handled by Paradrop rather than
# UCI configuration system.
WIFI_NONSTANDARD_MODES = set(["airshark"])


def isVirtual(ifname):
    """
    Test if an interface is a virtual one.

    FIXME: This just tests for the presence of certain strings in the interface
    name, so it is not very robust.
    """
    for marker in VIF_MARKERS:
        if marker in ifname:
            return True
    return False


def isWAN(ifname):
    """
    Test if an interface is a WAN interface.
    """
    pattern = re.compile(r"(\w+)\s+(\w+)*")
    routeList = pdos.readFile("/proc/net/route")
    for line in routeList:
        match = pattern.match(line)
        if match is not None and \
                match.group(1) == ifname and \
                match.group(2) == "00000000":
            return True
    return False


def isWireless(ifname):
    """
    Test if an interface is a wireless device.
    """
    check_path = "{}/{}/wireless".format(SYS_DIR, ifname)
    return pdos.exists(check_path)


def detectSystemDevices():
    """
    Detect devices on the system.

    The result is three lists stored in a dictionary.  The three lists are
    indexed by 'wan', 'wifi', and 'lan'.  Other devices may be supported by
    adding additional lists.

    Within each list, a device is represented by a dictionary.
    For all devices, the 'name' and 'mac' fields are defined.
    For WiFi devices, the 'phy' is defined in addition.
    Later, we may fill in more device information
    (e.g. what channels a WiFi card supports).
    """
    devices = dict()
    devices['wan'] = list()
    devices['wifi'] = list()
    devices['lan'] = list()

    for dev in listSystemDevices():
        devices[dev['type']].append(dev)
        del dev['type']

    return devices


def readSysFile(path):
    try:
        with open(path, 'r') as source:
            return source.read().strip()
    except:
        return None


def getMACAddress(ifname):
    path = "{}/{}/address".format(SYS_DIR, ifname)
    return readSysFile(path)


def getPhyMACAddress(phy):
    path = "{}/{}/macaddress".format(IEEE80211_DIR, phy)
    return readSysFile(path)


def getWirelessPhyName(ifname):
    path = "{}/{}/phy80211/name".format(SYS_DIR, ifname)
    return readSysFile(path)


class SysReader(object):
    PCI_BUS_ID = re.compile(r"\d+:\d+:\d+\.\d+")
    USB_BUS_ID = re.compile(r"\d+\-\d+:\d+\.\d+")

    def __init__(self, phy):
        self.phy = phy
        self.device_path = "{}/{}/device".format(IEEE80211_DIR, phy)

    def getDeviceId(self, default="????"):
        """
        Return the device ID for the device.

        This is a four-digit hexadecimal number.  For example, our Qualcomm
        802.11n chips have device ID 002a.
        """
        path = os.path.join(self.device_path, "device")
        device = readSysFile(path)
        if device is None:
            device = default
        return device

    def getSlotName(self, default=None):
        """
        Return the PCI/USB slot name for the device.

        Example: "pci/0000:04:00.0" or "usb/1-1:1.0"
        """
        path = os.path.join(self.device_path, "driver")
        for fname in os.listdir(path):
            match = SysReader.PCI_BUS_ID.match(fname)
            if match is not None:
                return "pci/" + fname

            match = SysReader.USB_BUS_ID.match(fname)
            if match is not None:
                return "usb/" + fname

    def getVendorId(self, default="????"):
        """
        Return the vendor ID for the device.

        This is a four-digit hexadecimal number.  For example, our Qualcomm
        802.11n chips have vendor ID 168c.
        """
        path = os.path.join(self.device_path, "vendor")
        vendor = readSysFile(path)
        if vendor is None:
            vendor = default
        return vendor


def listWiFiDevices():
    # Collect information about the physical devices (e.g. phy0 -> MAC address,
    # device type, PCI slot, etc.) and store as objects in a dictionary.
    devices = dict()
    try:
        for phy in pdos.listdir(IEEE80211_DIR):
            mac = getPhyMACAddress(phy)
            reader = SysReader(phy)

            devices[phy] = {
                'name': "wifi{}".format(mac.replace(':', '')),
                'type': 'wifi',
                'mac': mac,
                'phy': phy,
                'vendor': reader.getVendorId(),
                'device': reader.getDeviceId(),
                'slot': reader.getSlotName()
            }
    except OSError:
        # If we get an error here, it probably just means there are no WiFi
        # devices.
        pass

    # Collect a list of interfaces corresponding to each physical device
    # (e.g. phy0 -> wlan0, vwlan0.0000, etc.)
    interfaces = dict((dev, []) for dev in devices.keys())
    for ifname in pdos.listdir(SYS_DIR):
        try:<commit_msg>Added read_uevent method for future use.<commit_after>"""
Detect physical devices that can be used by chutes.

This module detects physical devices (for now just network interfaces) that can
be used by chutes.  This includes WAN interfaces for Internet connectivity and
WiFi interfaces which can host APs.

It also makes sure certain entries exist in the system UCI files for these
devices, for example "wifi-device" sections.  These are shared between chutes,
so they only need to be added when missing.
"""

import itertools
import operator
import os
import re
import subprocess

from paradrop.base.output import out
from paradrop.base import settings
from paradrop.lib.utils import datastruct, pdos, uci
from paradrop.core.config import uciutils

IEEE80211_DIR = "/sys/class/ieee80211"
SYS_DIR = "/sys/class/net"
EXCLUDE_IFACES = set(["lo"])

# Strings that identify a virtual interface.
VIF_MARKERS = [".", "veth"]

# Matches various ways of specifying WiFi devices (phy0, wlan0).
WIFI_DEV_REF = re.compile("([a-z]+)(\d+)")

# Set of wifi-interface mode values that are handled by Paradrop rather than
# UCI configuration system.
WIFI_NONSTANDARD_MODES = set(["airshark"])


def isVirtual(ifname):
    """
    Test if an interface is a virtual one.

    FIXME: This just tests for the presence of certain strings in the interface
    name, so it is not very robust.
    """
    for marker in VIF_MARKERS:
        if marker in ifname:
            return True
    return False


def isWAN(ifname):
    """
    Test if an interface is a WAN interface.
    """
    pattern = re.compile(r"(\w+)\s+(\w+)*")
    routeList = pdos.readFile("/proc/net/route")
    for line in routeList:
        match = pattern.match(line)
        if match is not None and \
                match.group(1) == ifname and \
                match.group(2) == "00000000":
            return True
    return False


def isWireless(ifname):
    """
    Test if an interface is a wireless device.
    """
    check_path = "{}/{}/wireless".format(SYS_DIR, ifname)
    return pdos.exists(check_path)


def detectSystemDevices():
    """
    Detect devices on the system.

    The result is three lists stored in a dictionary.  The three lists are
    indexed by 'wan', 'wifi', and 'lan'.  Other devices may be supported by
    adding additional lists.

    Within each list, a device is represented by a dictionary.
    For all devices, the 'name' and 'mac' fields are defined.
    For WiFi devices, the 'phy' is defined in addition.
    Later, we may fill in more device information
    (e.g. what channels a WiFi card supports).
    """
    devices = dict()
    devices['wan'] = list()
    devices['wifi'] = list()
    devices['lan'] = list()

    for dev in listSystemDevices():
        devices[dev['type']].append(dev)
        del dev['type']

    return devices


def readSysFile(path):
    try:
        with open(path, 'r') as source:
            return source.read().strip()
    except:
        return None


def getMACAddress(ifname):
    path = "{}/{}/address".format(SYS_DIR, ifname)
    return readSysFile(path)


def getPhyMACAddress(phy):
    path = "{}/{}/macaddress".format(IEEE80211_DIR, phy)
    return readSysFile(path)


def getWirelessPhyName(ifname):
    path = "{}/{}/phy80211/name".format(SYS_DIR, ifname)
    return readSysFile(path)


class SysReader(object):
    PCI_BUS_ID = re.compile(r"\d+:\d+:\d+\.\d+")
    USB_BUS_ID = re.compile(r"\d+\-\d+:\d+\.\d+")

    def __init__(self, phy):
        self.phy = phy
        self.device_path = "{}/{}/device".format(IEEE80211_DIR, phy)

    def getDeviceId(self, default="????"):
        """
        Return the device ID for the device.

        This is a four-digit hexadecimal number.  For example, our Qualcomm
        802.11n chips have device ID 002a.
        """
        path = os.path.join(self.device_path, "device")
        device = readSysFile(path)
        if device is None:
            device = default
        return device

    def getSlotName(self, default=None):
        """
        Return the PCI/USB slot name for the device.

        Example: "pci/0000:04:00.0" or "usb/1-1:1.0"
        """
        path = os.path.join(self.device_path, "driver")
        for fname in os.listdir(path):
            match = SysReader.PCI_BUS_ID.match(fname)
            if match is not None:
                return "pci/" + fname

            match = SysReader.USB_BUS_ID.match(fname)
            if match is not None:
                return "usb/" + fname

    def getVendorId(self, default="????"):
        """
        Return the vendor ID for the device.

        This is a four-digit hexadecimal number.  For example, our Qualcomm
        802.11n chips have vendor ID 168c.
        """
        path = os.path.join(self.device_path, "vendor")
        vendor = readSysFile(path)
        if vendor is None:
            vendor = default
        return vendor

    def read_uevent(self):
        """
        Read the device uevent file and return the contents as a dictionary.
        """
        result = dict()
        path = os.path.join(self.device_path, "uevent")
        with open(path, "r") as source:
            for line in source:
                key, value = line.split("=")
                result[key] = value
        return result


def listWiFiDevices():
    # Collect information about the physical devices (e.g. phy0 -> MAC address,
    # device type, PCI slot, etc.) and store as objects in a dictionary.
    devices = dict()
    try:
        for phy in pdos.listdir(IEEE80211_DIR):
            mac = getPhyMACAddress(phy)
            reader = SysReader(phy)

            devices[phy] = {
                'name': "wifi{}".format(mac.replace(':', '')),
                'type': 'wifi',
                'mac': mac,
                'phy': phy,
                'vendor': reader.getVendorId(),
                'device': reader.getDeviceId(),
                'slot': reader.getSlotName()
            }
    except OSError:
        # If we get an error here, it probably just means there are no WiFi
        # devices.
        pass

    # Collect a list of interfaces corresponding to each physical device
    # (e.g. phy0 -> wlan0, vwlan0.0000, etc.)
    interfaces = dict((dev, []) for dev in devices.keys())
    for ifname in pdos.listdir(SYS_DIR):
        try:
{'fraction_non_alphanumeric': 0.11063794938989435, 'fraction_numerical': 0.0129391532225043, 'mean_word_length': 2.37534549474848, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '667645', 'n_tokens_mistral': 3856, 'n_tokens_neox': 3613, 'n_words': 1225}