<commit_before>#
# This file is part of Plinth.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

"""
Helper functions for working with network manager.
"""

import collections
from django.utils.translation import ugettext_lazy as _
import logging
import socket
import struct
import subprocess
import uuid

from plinth.utils import import_from_gi
glib = import_from_gi('GLib', '2.0')
nm = import_from_gi('NM', '1.0')

logger = logging.getLogger(__name__)

CONNECTION_TYPE_NAMES = collections.OrderedDict([
    ('802-3-ethernet', _('Ethernet')),
    ('802-11-wireless', _('Wi-Fi')),
    ('pppoe', _('PPPoE'))
])


class ConnectionNotFound(Exception):
    """Network connection with a given name could not be found."""
    pass


class DeviceNotFound(Exception):
    """Network device for specified operation could not be found."""
    pass


def ipv4_string_to_int(address):
    """Return an integer equivalent of a string contain IPv4 address."""
    return struct.unpack('=I', socket.inet_aton(address))[0]


def ipv4_int_to_string(address_int):
    """Return an string equivalent of a integer IPv4 address."""
    return socket.inet_ntoa(struct.pack('=I', address_int))


def _callback(source_object, result, user_data):
    """Called when an operation is completed."""
    del source_object  # Unused
    del result  # Unused
    del user_data  # Unused


def _commit_callback(connection, error, data=None):
    """Called when the connection changes are committed."""
    del connection
    del error
    del data


def get_interface_list(device_type):
    """Get a list of network interface available on the system."""
    interfaces = {}
    for device in nm.Client.new(None).get_devices():
        if device.get_device_type() == device_type:
            interfaces[device.get_iface()] = device.get_hw_address()

    return interfaces


def get_status_from_connection(connection):
    """Return the current status of a connection."""
    status = collections.defaultdict(dict)

    status['id'] = connection.get_id()
    status['uuid'] = connection.get_uuid()
    status['type'] = connection.get_connection_type()
    status['zone'] = connection.get_setting_connection().get_zone()
    status['interface_name'] = connection.get_interface_name()

    status['ipv4']['method'] = connection.get_setting_ip4_config().get_method()
    status['ipv6']['method'] = connection.get_setting_ip6_config().get_method()

    if status['type'] == '802-11-wireless':
        setting_wireless = connection.get_setting_wireless()
        status['wireless']['ssid'] = setting_wireless.get_ssid().get_data()

    primary_connection = nm.Client.new(None).get_primary_connection()
    status['primary'] = (primary_connection.get_uuid() == connection.get_uuid())

    return status


def get_status_from_active_connection(connection):
    """Return the current status of an active connection."""
    status = collections.defaultdict(dict)

    status['state'] = connection.get_state().value_name
    status['ip4']['default'] = connection.get_default()
    status['ip6']['default'] = connection.get_default6()

    return status


def get_status_from_device(device):
    """Return a dictionary with current status of a network device."""
    if not device:
        return None

    status = collections.defaultdict(dict)

    ip4_config = device.get_ip4_config()
    if ip4_config:
        addresses = ip4_config.get_addresses()
        status['ip4']['addresses'] = [{'address': address.get_address(),
                                       'prefix': address.get_prefix()}
                                      for address in addresses]
        status['ip4']['gateway'] = ip4_config.get_gateway()
        status['ip4']['nameservers'] = ip4_config.get_nameservers()

    ip6_config = device.get_ip6_config()
    if ip6_config:
        addresses = ip6_config.get_addresses()
        status['ip6']['addresses'] = [{'address': address.get_address(),
                                       'prefix': address.get_prefix()}
                                      for address in addresses]
        status['ip6']['gateway'] = ip6_config.get_gateway()
        status['ip6']['nameservers'] = ip6_config.get_nameservers()

    status['type'] = device.get_type_description()
    status['description'] = device.get_description()
    status['hw_address'] = device.get_hw_address()
    status['interface_name'] = device.get_iface()
    status['state'] = device.get_state().value_nick
    status['state_reason'] = device.get_state_reason().value_nick

    if device.get_device_type() == nm.DeviceType.WIFI:
        status['wireless']['bitrate'] = device.get_bitrate() / 1000
        status['wireless']['mode'] = device.get_mode().value_nick

    if device.get_device_type() == nm.DeviceType.ETHERNET:
        status['ethernet']['speed'] = device.get_speed()
        status['ethernet']['carrier'] = device.get_carrier()

    return status


def get_status_from_wifi_access_point(device, ssid):
    """Return the current status of an access point."""
    status = {}

    if not ssid or not device:
        return status

    for access_point in device.get_access_points():
        if access_point.get_ssid().get_data() == ssid:
            status['strength'] = access_point.get_strength()
            frequency = access_point.get_frequency()
            status['channel'] = _get_wifi_channel_from_frequency(frequency)
            break

    return status


def _get_wifi_channel_from_frequency(frequency):
    """Get the wifi channel form a particular SSID"""
    # TODO: Hard coded list of wifi frequencys and their corresponding
    # channel numbers.  Search for a better solution!  Even 5GHz is
    # not included yet.  Only the plain frequency will show up on 5GHz
    # AP's.
    channel_map = {2412: 1, 2417: 2, 2422: 3, 2427: 4, 2432: 5, 2437: 6,
                   2442: 7, 2447: 8, 2452: 9, 2457: 10, 2462: 11}
    try:
        return channel_map[frequency]
    except KeyError:
        return str(frequency / 1000) + 'GHz'


def get_first_ip_address_from_connection(connection):
    """Return the first IP address of a connection setting.

    XXX: Work around a bug in NetworkManager/Python GI.  Remove after
    the bug if fixed.
    https://bugzilla.gnome.org/show_bug.cgi?id=756380.
    """
    command = ['nmcli', '--terse', '--mode', 'tabular', '--fields',
               'ipv4.addresses', 'connection', 'show', connection.get_uuid()]

    output = subprocess.check_output(command).decode()
    first = output.strip().split(', ')[0]
    if not first:
        return None, None

    ip_address, prefix = first.split('/')

    netmask = nm.utils_ip4_prefix_to_netmask(int(prefix))
    return ip_address, ipv4_int_to_string(netmask)


def get_connection_list():
    """Get a list of active and available connections."""
    active_uuids = []
    client = nm.Client.new(None)
    for connection in client.get_active_connections():
        active_uuids.append(connection.get_uuid())

    connections = []
    for connection in client.get_connections():
        # Display a friendly type name if known.
        connection_type = connection.get_connection_type()
        connection_type_name = CONNECTION_TYPE_NAMES.get(connection_type,
                                                         connection_type)

        settings_connection = connection.get_setting_connection()
        zone = settings_connection.get_zone()

        connection.get_interface_name()

        connections.append({
            'name': connection.get_id(),
            'uuid': connection.get_uuid(),
            'interface_name': connection.get_interface_name(),
            'type': connection_type,
            'type_name': connection_type_name,
            'is_active': connection.get_uuid() in active_uuids,
            'zone': zone,
        })
    connections.sort(key=lambda connection: connection['is_active'],
                     reverse=True)
    return connections


def get_connection(connection_uuid):
    """Return connection with matching uuid.

    Raise ConnectionNotFound if a connection with that uuid is not
    found.
    """
    client = nm.Client.new(None)
    connection = client.get_connection_by_uuid(connection_uuid)
    if not connection:
        raise ConnectionNotFound(connection_uuid)

    return connection


def get_active_connection(connection_uuid):
    """Return active connection with matching UUID.

    Raise ConnectionNotFound if a connection with that uuid is not
    found.
    """
    connections = nm.Client.new(None).get_active_connections()
    connections = {connection.get_uuid(): connection
                   for connection in connections}
    try:
        return connections[connection_uuid]
    except KeyError:
        raise ConnectionNotFound(connection_uuid)


def get_device_by_interface_name(interface_name):
    """Return a device by interface name."""
    return nm.Client.new(None).get_device_by_iface(interface_name)


def _update_common_settings(connection, connection_uuid, name, type_,
                            interface, zone):
    """Create/edit basic settings for network manager connections.

    Return newly created connection object if connection is None.
    """
    if not connection:
        connection = nm.SimpleConnection.new()

    settings = connection.get_setting_connection()
    if not settings:
        settings = nm.SettingConnection.new()
        connection.add_setting(settings)

    settings.set_property(nm.SETTING_CONNECTION_UUID, connection_uuid)
    settings.set_property(nm.SETTING_CONNECTION_ID, name)
    settings.set_property(nm.SETTING_CONNECTION_TYPE, type_)
    settings.set_property(nm.SETTING_CONNECTION_INTERFACE_NAME, interface)
    settings.set_property(nm.SETTING_CONNECTION_ZONE, zone)

    return connection


def _update_ipv4_settings(connection, ipv4_method, ipv4_address, ipv4_netmask,
                          ipv4_gateway, ipv4_dns, ipv4_second_dns):
    """Edit IPv4 settings for network manager connections."""
    settings = nm.SettingIP4Config.new()
    connection.add_setting(settings)

    settings.set_property(nm.SETTING_IP_CONFIG_METHOD, ipv4_method)
    if ipv4_method == nm.SETTING_IP4_CONFIG_METHOD_MANUAL and ipv4_address:
        ipv4_address_int = ipv4_string_to_int(ipv4_address)
        if not ipv4_netmask:
            ipv4_netmask_int = nm.utils_ip4_get_default_prefix(
                               ipv4_address_int)

        ipv4_netmask_int = nm.utils_ip4_netmask_to_prefix(
                           ipv4_string_to_int(ipv4_netmask))
        address = nm.IPAddress.new(socket.AF_INET, ipv4_address,
                                   ipv4_netmask_int)
        settings.add_address(address)

        if ipv4_dns:
            settings.add_dns(ipv4_dns)
        if ipv4_second_dns:
            settings.add_dns(ipv4_second_dns)

        if not ipv4_gateway:
            settings.set_property(nm.SETTING_IP_CONFIG_GATEWAY, '0.0.0.0')
        else:
            settings.set_property(nm.SETTING_IP_CONFIG_GATEWAY, ipv4_gateway)
    else:
        if ipv4_dns or ipv4_second_dns:
            settings.set_property(nm.SETTING_IP_CONFIG_IGNORE_AUTO_DNS, True)

        if ipv4_dns:
            settings.add_dns(ipv4_dns)
        if ipv4_second_dns:
            settings.add_dns(ipv4_second_dns)


def _update_ethernet_settings(connection, connection_uuid, name, interface,
                              zone, ipv4_method, ipv4_address, ipv4_netmask,
                              ipv4_gateway, ipv4_dns, ipv4_second_dns):
    """Create/edit ethernet settings for network manager connections."""
    type_ = '802-3-ethernet'

    connection = _update_common_settings(connection, connection_uuid, name,
                                         type_, interface, zone)
    _update_ipv4_settings(connection, ipv4_method, ipv4_address, ipv4_netmask,<commit_msg>networks: Fix issue with setting default netmask<commit_after>#
# This file is part of Plinth.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

"""
Helper functions for working with network manager.
"""

import collections
from django.utils.translation import ugettext_lazy as _
import logging
import socket
import struct
import subprocess
import uuid

from plinth.utils import import_from_gi
glib = import_from_gi('GLib', '2.0')
nm = import_from_gi('NM', '1.0')

logger = logging.getLogger(__name__)

CONNECTION_TYPE_NAMES = collections.OrderedDict([
    ('802-3-ethernet', _('Ethernet')),
    ('802-11-wireless', _('Wi-Fi')),
    ('pppoe', _('PPPoE'))
])


class ConnectionNotFound(Exception):
    """Network connection with a given name could not be found."""
    pass


class DeviceNotFound(Exception):
    """Network device for specified operation could not be found."""
    pass


def ipv4_string_to_int(address):
    """Return an integer equivalent of a string contain IPv4 address."""
    return struct.unpack('=I', socket.inet_aton(address))[0]


def ipv4_int_to_string(address_int):
    """Return an string equivalent of a integer IPv4 address."""
    return socket.inet_ntoa(struct.pack('=I', address_int))


def _callback(source_object, result, user_data):
    """Called when an operation is completed."""
    del source_object  # Unused
    del result  # Unused
    del user_data  # Unused


def _commit_callback(connection, error, data=None):
    """Called when the connection changes are committed."""
    del connection
    del error
    del data


def get_interface_list(device_type):
    """Get a list of network interface available on the system."""
    interfaces = {}
    for device in nm.Client.new(None).get_devices():
        if device.get_device_type() == device_type:
            interfaces[device.get_iface()] = device.get_hw_address()

    return interfaces


def get_status_from_connection(connection):
    """Return the current status of a connection."""
    status = collections.defaultdict(dict)

    status['id'] = connection.get_id()
    status['uuid'] = connection.get_uuid()
    status['type'] = connection.get_connection_type()
    status['zone'] = connection.get_setting_connection().get_zone()
    status['interface_name'] = connection.get_interface_name()

    status['ipv4']['method'] = connection.get_setting_ip4_config().get_method()
    status['ipv6']['method'] = connection.get_setting_ip6_config().get_method()

    if status['type'] == '802-11-wireless':
        setting_wireless = connection.get_setting_wireless()
        status['wireless']['ssid'] = setting_wireless.get_ssid().get_data()

    primary_connection = nm.Client.new(None).get_primary_connection()
    status['primary'] = (primary_connection.get_uuid() == connection.get_uuid())

    return status


def get_status_from_active_connection(connection):
    """Return the current status of an active connection."""
    status = collections.defaultdict(dict)

    status['state'] = connection.get_state().value_name
    status['ip4']['default'] = connection.get_default()
    status['ip6']['default'] = connection.get_default6()

    return status


def get_status_from_device(device):
    """Return a dictionary with current status of a network device."""
    if not device:
        return None

    status = collections.defaultdict(dict)

    ip4_config = device.get_ip4_config()
    if ip4_config:
        addresses = ip4_config.get_addresses()
        status['ip4']['addresses'] = [{'address': address.get_address(),
                                       'prefix': address.get_prefix()}
                                      for address in addresses]
        status['ip4']['gateway'] = ip4_config.get_gateway()
        status['ip4']['nameservers'] = ip4_config.get_nameservers()

    ip6_config = device.get_ip6_config()
    if ip6_config:
        addresses = ip6_config.get_addresses()
        status['ip6']['addresses'] = [{'address': address.get_address(),
                                       'prefix': address.get_prefix()}
                                      for address in addresses]
        status['ip6']['gateway'] = ip6_config.get_gateway()
        status['ip6']['nameservers'] = ip6_config.get_nameservers()

    status['type'] = device.get_type_description()
    status['description'] = device.get_description()
    status['hw_address'] = device.get_hw_address()
    status['interface_name'] = device.get_iface()
    status['state'] = device.get_state().value_nick
    status['state_reason'] = device.get_state_reason().value_nick

    if device.get_device_type() == nm.DeviceType.WIFI:
        status['wireless']['bitrate'] = device.get_bitrate() / 1000
        status['wireless']['mode'] = device.get_mode().value_nick

    if device.get_device_type() == nm.DeviceType.ETHERNET:
        status['ethernet']['speed'] = device.get_speed()
        status['ethernet']['carrier'] = device.get_carrier()

    return status


def get_status_from_wifi_access_point(device, ssid):
    """Return the current status of an access point."""
    status = {}

    if not ssid or not device:
        return status

    for access_point in device.get_access_points():
        if access_point.get_ssid().get_data() == ssid:
            status['strength'] = access_point.get_strength()
            frequency = access_point.get_frequency()
            status['channel'] = _get_wifi_channel_from_frequency(frequency)
            break

    return status


def _get_wifi_channel_from_frequency(frequency):
    """Get the wifi channel form a particular SSID"""
    # TODO: Hard coded list of wifi frequencys and their corresponding
    # channel numbers.  Search for a better solution!  Even 5GHz is
    # not included yet.  Only the plain frequency will show up on 5GHz
    # AP's.
    channel_map = {2412: 1, 2417: 2, 2422: 3, 2427: 4, 2432: 5, 2437: 6,
                   2442: 7, 2447: 8, 2452: 9, 2457: 10, 2462: 11}
    try:
        return channel_map[frequency]
    except KeyError:
        return str(frequency / 1000) + 'GHz'


def get_first_ip_address_from_connection(connection):
    """Return the first IP address of a connection setting.

    XXX: Work around a bug in NetworkManager/Python GI.  Remove after
    the bug if fixed.
    https://bugzilla.gnome.org/show_bug.cgi?id=756380.
    """
    command = ['nmcli', '--terse', '--mode', 'tabular', '--fields',
               'ipv4.addresses', 'connection', 'show', connection.get_uuid()]

    output = subprocess.check_output(command).decode()
    first = output.strip().split(', ')[0]
    if not first:
        return None, None

    ip_address, prefix = first.split('/')

    netmask = nm.utils_ip4_prefix_to_netmask(int(prefix))
    return ip_address, ipv4_int_to_string(netmask)


def get_connection_list():
    """Get a list of active and available connections."""
    active_uuids = []
    client = nm.Client.new(None)
    for connection in client.get_active_connections():
        active_uuids.append(connection.get_uuid())

    connections = []
    for connection in client.get_connections():
        # Display a friendly type name if known.
        connection_type = connection.get_connection_type()
        connection_type_name = CONNECTION_TYPE_NAMES.get(connection_type,
                                                         connection_type)

        settings_connection = connection.get_setting_connection()
        zone = settings_connection.get_zone()

        connection.get_interface_name()

        connections.append({
            'name': connection.get_id(),
            'uuid': connection.get_uuid(),
            'interface_name': connection.get_interface_name(),
            'type': connection_type,
            'type_name': connection_type_name,
            'is_active': connection.get_uuid() in active_uuids,
            'zone': zone,
        })
    connections.sort(key=lambda connection: connection['is_active'],
                     reverse=True)
    return connections


def get_connection(connection_uuid):
    """Return connection with matching uuid.

    Raise ConnectionNotFound if a connection with that uuid is not
    found.
    """
    client = nm.Client.new(None)
    connection = client.get_connection_by_uuid(connection_uuid)
    if not connection:
        raise ConnectionNotFound(connection_uuid)

    return connection


def get_active_connection(connection_uuid):
    """Return active connection with matching UUID.

    Raise ConnectionNotFound if a connection with that uuid is not
    found.
    """
    connections = nm.Client.new(None).get_active_connections()
    connections = {connection.get_uuid(): connection
                   for connection in connections}
    try:
        return connections[connection_uuid]
    except KeyError:
        raise ConnectionNotFound(connection_uuid)


def get_device_by_interface_name(interface_name):
    """Return a device by interface name."""
    return nm.Client.new(None).get_device_by_iface(interface_name)


def _update_common_settings(connection, connection_uuid, name, type_,
                            interface, zone):
    """Create/edit basic settings for network manager connections.

    Return newly created connection object if connection is None.
    """
    if not connection:
        connection = nm.SimpleConnection.new()

    settings = connection.get_setting_connection()
    if not settings:
        settings = nm.SettingConnection.new()
        connection.add_setting(settings)

    settings.set_property(nm.SETTING_CONNECTION_UUID, connection_uuid)
    settings.set_property(nm.SETTING_CONNECTION_ID, name)
    settings.set_property(nm.SETTING_CONNECTION_TYPE, type_)
    settings.set_property(nm.SETTING_CONNECTION_INTERFACE_NAME, interface)
    settings.set_property(nm.SETTING_CONNECTION_ZONE, zone)

    return connection


def _update_ipv4_settings(connection, ipv4_method, ipv4_address, ipv4_netmask,
                          ipv4_gateway, ipv4_dns, ipv4_second_dns):
    """Edit IPv4 settings for network manager connections."""
    settings = nm.SettingIP4Config.new()
    connection.add_setting(settings)

    settings.set_property(nm.SETTING_IP_CONFIG_METHOD, ipv4_method)
    if ipv4_method == nm.SETTING_IP4_CONFIG_METHOD_MANUAL and ipv4_address:
        ipv4_address_int = ipv4_string_to_int(ipv4_address)

        if not ipv4_netmask:
            ipv4_netmask_int = nm.utils_ip4_get_default_prefix(
                ipv4_address_int)
        else:
            ipv4_netmask_int = nm.utils_ip4_netmask_to_prefix(
                ipv4_string_to_int(ipv4_netmask))

        address = nm.IPAddress.new(socket.AF_INET, ipv4_address,
                                   ipv4_netmask_int)
        settings.add_address(address)

        if not ipv4_gateway:
            settings.set_property(nm.SETTING_IP_CONFIG_GATEWAY, '0.0.0.0')
        else:
            settings.set_property(nm.SETTING_IP_CONFIG_GATEWAY, ipv4_gateway)
    else:
        if ipv4_dns or ipv4_second_dns:
            settings.set_property(nm.SETTING_IP_CONFIG_IGNORE_AUTO_DNS, True)

    if ipv4_dns:
        settings.add_dns(ipv4_dns)

    if ipv4_second_dns:
        settings.add_dns(ipv4_second_dns)


def _update_ethernet_settings(connection, connection_uuid, name, interface,
                              zone, ipv4_method, ipv4_address, ipv4_netmask,
                              ipv4_gateway, ipv4_dns, ipv4_second_dns):
    """Create/edit ethernet settings for network manager connections."""
    type_ = '802-3-ethernet'

    connection = _update_common_settings(connection, connection_uuid, name,
                                         type_, interface, zone)
    _update_ipv4_settings(connection, ipv4_method, ipv4_address, ipv4_netmask,
{'fraction_non_alphanumeric': 0.10140481188438559, 'fraction_numerical': 0.014532536735023414, 'mean_word_length': 3.2052283143778646, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3171199', 'n_tokens_mistral': 7423, 'n_tokens_neox': 7147, 'n_words': 1726}