Файловый менеджер - Редактировать - /var/www/html/cli.zip
Ðазад
PK ! enabled_by_maint/opcachenu �[��� PK ! enabled_by_maint/zipnu �[��� PK ! enabled_by_maint/sysvsemnu �[��� PK ! enabled_by_maint/curlnu �[��� PK ! enabled_by_maint/iconvnu �[��� PK ! enabled_by_maint/readlinenu �[��� PK ! enabled_by_maint/sysvmsgnu �[��� PK ! enabled_by_maint/pdo_mysqlnu �[��� PK ! enabled_by_maint/exifnu �[��� PK ! enabled_by_maint/shmopnu �[��� PK ! enabled_by_maint/gettextnu �[��� PK ! enabled_by_maint/ftpnu �[��� PK ! enabled_by_maint/pdonu �[��� PK ! enabled_by_maint/mysqlndnu �[��� PK ! enabled_by_maint/fileinfonu �[��� PK ! enabled_by_maint/pharnu �[��� PK ! enabled_by_maint/simplexmlnu �[��� PK ! enabled_by_maint/ctypenu �[��� PK ! enabled_by_maint/xmlwriternu �[��� PK ! enabled_by_maint/domnu �[��� PK ! enabled_by_maint/socketsnu �[��� PK ! enabled_by_maint/gdnu �[��� PK ! enabled_by_maint/ffinu �[��� PK ! enabled_by_maint/mbstringnu �[��� PK ! enabled_by_maint/sysvshmnu �[��� PK ! enabled_by_maint/mysqlinu �[��� PK ! enabled_by_maint/calendarnu �[��� PK ! enabled_by_maint/posixnu �[��� PK ! enabled_by_maint/xmlnu �[��� PK ! enabled_by_maint/xmlreadernu �[��� PK ! enabled_by_maint/xslnu �[��� PK ! enabled_by_maint/tokenizernu �[��� PK ! enabled_by_maint/intlnu �[��� PK ! �8�� � ovs.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2020 Canonical, Ltd. # Author: Lukas 'slyon' Märdian <lukas.maerdian@canonical.com> # # 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; version 3. # # 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 logging import os import subprocess import re from netplan.cli.utils import systemctl_is_active OPENVSWITCH_OVS_VSCTL = '/usr/bin/ovs-vsctl' OPENVSWITCH_OVSDB_SERVER_UNIT = 'ovsdb-server.service' # Defaults for non-optional settings, as defined here: # http://www.openvswitch.org/ovs-vswitchd.conf.db.5.pdf DEFAULTS = { # Mandatory columns: 'mcast_snooping_enable': 'false', 'rstp_enable': 'false', } GLOBALS = { # Global commands: 'set-ssl': ('del-ssl', 'get-ssl'), 'set-fail-mode': ('del-fail-mode', 'get-fail-mode'), 'set-controller': ('del-controller', 'get-controller'), } class OvsDbServerNotRunning(Exception): pass def _del_col(type, iface, column, value): """Cleanup values from a column (i.e. "column=value")""" default = DEFAULTS.get(column) if default is None: # removes the exact value only if it was set by netplan subprocess.check_call([OPENVSWITCH_OVS_VSCTL, 'remove', type, iface, column, value]) elif default and default != value: # reset to default, if its not the default already subprocess.check_call([OPENVSWITCH_OVS_VSCTL, 'set', type, iface, '%s=%s' % (column, default)]) def _del_dict(type, iface, column, key, value): """Cleanup values from a dictionary (i.e. "column:key=value")""" # removes the exact value only if it was set by netplan subprocess.check_call([OPENVSWITCH_OVS_VSCTL, 'remove', type, iface, column, key, _escape_colon(value)]) # for ovsdb remove: column key's value can not contain bare ':', need to escape with '\' def _escape_colon(literal): return re.sub(r'([^\\]):', r'\g<1>\:', literal) def _del_global(type, iface, key, value): """Cleanup commands from the global namespace""" del_cmd, get_cmd = GLOBALS.get(key, (None, None)) if del_cmd == 'del-ssl': iface = None if del_cmd: args_get = [OPENVSWITCH_OVS_VSCTL, get_cmd] args_del = [OPENVSWITCH_OVS_VSCTL, del_cmd] if iface: args_get.append(iface) args_del.append(iface) # Check the current value of a global command and compare it to the tag-value, e.g.: # * get-ssl: netplan/global/set-ssl=/private/key.pem,/another/cert.pem,/some/ca-cert.pem # Private key: /private/key.pem # Certificate: /another/cert.pem # CA Certificate: /some/ca-cert.pem # Bootstrap: false # * get-fail-mode: netplan/global/set-fail-mode=secure # secure # * get-controller: netplan/global/set-controller=tcp:127.0.0.1:1337,unix:/some/socket # tcp:127.0.0.1:1337 # unix:/some/socket out = subprocess.check_output(args_get, text=True) # Clean it only if the exact same value(s) were set by netplan. # Don't touch it if other values were set by another integration. if all(item in out for item in value.split(',')): subprocess.check_call(args_del) else: raise Exception('Reset command unknown for:', key) def clear_setting(type, iface, setting, value): """Check if this setting is in a dict or a colum and delete accordingly""" split = setting.split('/', 2) col = split[1] if col == 'global' and len(split) > 2: _del_global(type, iface, split[2], value) elif len(split) > 2: _del_dict(type, iface, split[1], split[2], value) else: _del_col(type, iface, split[1], value) # Cleanup the tag itself (i.e. "netplan/column[/key]") subprocess.check_call([OPENVSWITCH_OVS_VSCTL, 'remove', type, iface, 'external-ids', setting]) def is_ovs_interface(iface, np_interface_dict): assert isinstance(np_interface_dict, dict) np_def = np_interface_dict.get(iface, None) return np_def and np_def.backend == 'OpenVSwitch' def apply_ovs_cleanup(config_manager, ovs_old, ovs_current): # pragma: nocover (covered in autopkgtest) """ Query OpenVSwitch state through 'ovs-vsctl' and filter for netplan=true tagged ports/bonds and bridges. Delete interfaces which are not defined in the current configuration. Also filter for individual settings tagged netplan/<column>[/<key]=value in external-ids and clear them if they have been set by netplan. """ if not systemctl_is_active(OPENVSWITCH_OVSDB_SERVER_UNIT): raise OvsDbServerNotRunning('{} is not running'.format(OPENVSWITCH_OVSDB_SERVER_UNIT)) config_manager.parse() ovs_ifaces = set() for i in config_manager.all_defs.keys(): if (is_ovs_interface(i, config_manager.all_defs)): ovs_ifaces.add(i) # Tear down old OVS interfaces, not defined in the current config. # Use 'del-br' on the Interface table, to delete any netplan created VLAN fake bridges. # Use 'del-bond-iface' on the Interface table, to delete netplan created patch port interfaces if os.path.isfile(OPENVSWITCH_OVS_VSCTL): # Step 1: Delete all interfaces, which are not part of the current OVS config for t in (('Port', 'del-port'), ('Bridge', 'del-br'), ('Interface', 'del-br')): out = subprocess.check_output([OPENVSWITCH_OVS_VSCTL, '--columns=name,external-ids', '-f', 'csv', '-d', 'bare', '--no-headings', 'list', t[0]], text=True) for line in out.splitlines(): if 'netplan=true' in line: iface = line.split(',')[0] # Skip cleanup if this OVS interface is part of the current netplan OVS config if iface in ovs_ifaces: continue if t[0] == 'Interface' and subprocess.run([OPENVSWITCH_OVS_VSCTL, 'br-exists', iface]).returncode > 0: subprocess.check_call([OPENVSWITCH_OVS_VSCTL, '--if-exists', 'del-bond-iface', iface]) else: subprocess.check_call([OPENVSWITCH_OVS_VSCTL, '--if-exists', t[1], iface]) # Step 2: Clean up the settings of the remaining interfaces for t in ('Port', 'Bridge', 'Interface', 'Open_vSwitch', 'Controller'): cols = 'name,external-ids' if t == 'Open_vSwitch': cols = 'external-ids' elif t == 'Controller': cols = '_uuid,external-ids' # handle _uuid as if it would be the iface 'name' out = subprocess.check_output([OPENVSWITCH_OVS_VSCTL, '--columns=%s' % cols, '-f', 'csv', '-d', 'bare', '--no-headings', 'list', t], text=True) for line in out.splitlines(): if 'netplan/' in line: iface = '.' extids = line if t != 'Open_vSwitch': iface, extids = line.split(',', 1) # Check each line (interface) if it contains any netplan tagged settings, e.g.: # ovs0,"iface-id=myhostname netplan=true netplan/external-ids/iface-id=myhostname" # ovs1,"netplan=true netplan/global/set-fail-mode=standalone netplan/mcast_snooping_enable=false" for entry in extids.strip('"').split(' '): if entry.startswith('netplan/') and '=' in entry: setting, val = entry.split('=', 1) clear_setting(t, iface, setting, val) # Show the warning only if we are or have been working with OVS definitions elif ovs_old or ovs_current: logging.warning('ovs-vsctl is missing, cannot tear down old OpenVSwitch interfaces') PK ! ��](�F �F sriov.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2020-2022 Canonical, Ltd. # Author: Łukasz 'sil2100' Zemczak <lukasz.zemczak@canonical.com> # Author: Lukas Märdian <slyon@ubuntu.com> # # 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; version 3. # # 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 logging import os import subprocess import typing from collections import defaultdict import netplan.cli.utils as utils import netplan.libnetplan as libnetplan from netplan.configmanager import ConfigurationError import netifaces # PCIDevice class originates from mlnx_switchdev_mode/sriovify.py # Copyright 2019 Canonical Ltd, Apache License, Version 2.0 # https://github.com/openstack-charmers/mlnx-switchdev-mode class PCIDevice(object): """Helper class for interaction with a PCI device""" def __init__(self, pci_addr: str): """Initialise a new PCI device handler :param pci_addr: PCI address of device :type: str """ self.pci_addr = pci_addr @property def sys(self) -> str: """sysfs path (can be overridden for testing) :return: full path to /sys filesystem :rtype: str """ return "/sys" @property def path(self) -> str: """/sys path for PCI device :return: full path to PCI device in /sys filesystem :rtype: str """ return os.path.join(self.sys, "bus/pci/devices", self.pci_addr) def subpath(self, subpath: str) -> str: """/sys subpath helper for PCI device :param subpath: subpath to construct path for :type: str :return: self.path + subpath :rtype: str """ return os.path.join(self.path, subpath) @property def driver(self) -> str: """Kernel driver for PCI device :return: kernel driver in use for device :rtype: str """ driver = '' if os.path.exists(self.subpath("driver")): driver = os.path.basename(os.readlink(self.subpath("driver"))) return driver @property def bound(self) -> bool: """Determine if device is bound to a kernel driver :return: whether device is bound to a kernel driver :rtype: bool """ return os.path.exists(self.subpath("driver")) @property def is_pf(self) -> bool: """Determine if device is a SR-IOV Physical Function :return: whether device is a PF :rtype: bool """ return os.path.exists(self.subpath("sriov_numvfs")) @property def is_vf(self) -> bool: """Determine if device is a SR-IOV Virtual Function :return: whether device is a VF :rtype: bool """ return os.path.exists(self.subpath("physfn")) @property def vf_addrs(self) -> list: """List Virtual Function addresses associated with a Physical Function :return: List of PCI addresses of Virtual Functions :rtype: list[str] """ vf_addrs = [] i = 0 while True: try: vf_addrs.append( os.path.basename( os.readlink(self.subpath("virtfn{}".format(i))) ) ) except FileNotFoundError: break i += 1 return vf_addrs @property def vfs(self) -> list: """List Virtual Function associated with a Physical Function :return: List of PCI devices of Virtual Functions :rtype: list[PCIDevice] """ return [PCIDevice(addr) for addr in self.vf_addrs] def devlink_set(self, obj_name: str, prop: str, value: str): """Set devlink options for the PCI device :param obj_name: devlink object to set options on :type: str :param prop: property to set :type: str :param value: value to set for property :type: str """ subprocess.check_call( [ "/sbin/devlink", "dev", obj_name, "set", "pci/{}".format(self.pci_addr), prop, value, ] ) def __str__(self) -> str: """String represenation of object :return: PCI address of string :rtype: str """ return self.pci_addr def bind_vfs(vfs: typing.Iterable[PCIDevice], driver): """Bind unbound VFs to driver.""" bound_vfs = [] for vf in vfs: if not vf.bound: with open("/sys/bus/pci/drivers/{}/bind".format(driver), "wt") as f: f.write(vf.pci_addr) bound_vfs.append(vf) return bound_vfs def unbind_vfs(vfs: typing.Iterable[PCIDevice], driver) -> typing.Iterable[PCIDevice]: """Unbind bound VFs from driver.""" unbound_vfs = [] for vf in vfs: if vf.bound: with open("/sys/bus/pci/drivers/{}/unbind".format(driver), "wt") as f: f.write(vf.pci_addr) unbound_vfs.append(vf) return unbound_vfs def _get_target_interface(interfaces, np_state, pf_link, pfs): if pf_link not in pfs: # handle the match: syntax, get the actual device name pf_dev = np_state[pf_link] if pf_dev.has_match: # now here it's a bit tricky set_name = pf_dev.set_name if set_name and set_name in interfaces: # if we had a match: stanza and set-name: this means we should # assume that, if found, the interface has already been # renamed - use the new name pfs[pf_link] = set_name else: for interface in interfaces: if not pf_dev.match_interface( itf_name=interface, itf_driver=utils.get_interface_driver_name(interface), itf_mac=utils.get_interface_macaddress(interface)): continue # we have a matching PF # store the matching interface in the dictionary of # active PFs, but error out if we matched more than one if pf_link in pfs: raise ConfigurationError('matched more than one interface for a PF device: %s' % pf_link) pfs[pf_link] = interface else: # no match field, assume entry name is the interface name if pf_link in interfaces: pfs[pf_link] = pf_link return pfs.get(pf_link, None) def _get_pci_slot_name(netdev): """ Read PCI slot name for given interface name """ uevent_path = os.path.join('/sys/class/net', netdev, 'device/uevent') try: with open(uevent_path) as f: pci_slot_name = None for line in f.readlines(): line = line.strip() if line.startswith('PCI_SLOT_NAME='): pci_slot_name = line.split('=', 2)[1] return pci_slot_name except IOError as e: raise RuntimeError('failed parsing PCI slot name for %s: %s' % (netdev, str(e))) def get_vf_count_and_functions(interfaces, np_state, vf_counts, vfs, pfs): """ Go through the list of netplan ethernet devices and identify which are PFs and VFs, matching the former with actual networking interfaces. Count how many VFs each PF will need. """ for nid, netdef in np_state.ethernets.items(): if netdef.sriov_link and _get_target_interface(interfaces, np_state, netdef.sriov_link.id, pfs): vfs[nid] = None try: count = netdef.vf_count except libnetplan.LibNetplanException as e: raise ConfigurationError(*e.args) if count == 0: continue pf = _get_target_interface(interfaces, np_state, nid, pfs) if pf: vf_counts[pf] = count def set_numvfs_for_pf(pf, vf_count): """ Allocate the required number of VFs for the selected PF. """ if vf_count > 256: raise ConfigurationError( 'cannot allocate more VFs for PF %s than the SR-IOV maximum: %s > 256' % (pf, vf_count)) devdir = os.path.join('/sys/class/net', pf, 'device') numvfs_path = os.path.join(devdir, 'sriov_numvfs') totalvfs_path = os.path.join(devdir, 'sriov_totalvfs') try: with open(totalvfs_path) as f: vf_max = int(f.read().strip()) except IOError as e: raise RuntimeError('failed parsing sriov_totalvfs for %s: %s' % (pf, str(e))) except ValueError: raise RuntimeError('invalid sriov_totalvfs value for %s' % pf) if vf_count > vf_max: raise ConfigurationError( 'cannot allocate more VFs for PF %s than supported: %s > %s (sriov_totalvfs)' % (pf, vf_count, vf_max)) try: with open(numvfs_path, 'w') as f: f.write(str(vf_count)) except IOError as e: bail = True if e.errno == 16: # device or resource busy logging.warning('device or resource busy while setting sriov_numvfs for %s, trying workaround' % pf) try: # doing this in two open/close sequences so that # it's as close to writing via shell as possible with open(numvfs_path, 'w') as f: f.write('0') with open(numvfs_path, 'w') as f: f.write(str(vf_count)) except IOError as e_inner: e = e_inner else: bail = False if bail: raise RuntimeError('failed setting sriov_numvfs to %s for %s: %s' % (vf_count, pf, str(e))) return True def perform_hardware_specific_quirks(pf): """ Perform any hardware-specific quirks for the given SR-IOV device to make sure all the VF-count changes are applied. """ devdir = os.path.join('/sys/class/net', pf, 'device') try: with open(os.path.join(devdir, 'vendor')) as f: device_id = f.read().strip()[2:] with open(os.path.join(devdir, 'device')) as f: vendor_id = f.read().strip()[2:] except IOError as e: raise RuntimeError('could not determine vendor and device ID of %s: %s' % (pf, str(e))) combined_id = ':'.join([vendor_id, device_id]) quirk_devices = () # TODO: add entries to the list if combined_id in quirk_devices: # pragma: nocover (empty quirk_devices) # some devices need special handling, so this is the place # Currently this part is empty, but has been added as a preemptive # measure, as apparently a lot of SR-IOV cards have issues with # dynamically allocating VFs. Some cards seem to require a full # kernel module reload cycle after changing the sriov_numvfs value # for the changes to come into effect. # Any identified card/vendor can then be special-cased here, if # needed. pass def apply_vlan_filter_for_vf(pf, vf, vlan_name, vlan_id, prefix='/'): """ Apply the hardware VLAN filtering for the selected VF. """ # this is more complicated, because to do this, we actually need to have # the vf index - just knowing the vf interface name is not enough vf_index = None # the prefix argument is here only for unit testing purposes vf_devdir = os.path.join(prefix, 'sys/class/net', vf, 'device') vf_dev_id = os.path.basename(os.readlink(vf_devdir)) pf_devdir = os.path.join(prefix, 'sys/class/net', pf, 'device') for f in os.listdir(pf_devdir): if 'virtfn' in f: dev_path = os.path.join(pf_devdir, f) dev_id = os.path.basename(os.readlink(dev_path)) if dev_id == vf_dev_id: vf_index = f[6:] break if not vf_index: raise RuntimeError( 'could not determine the VF index for %s while configuring vlan %s' % (vf, vlan_name)) # now, create the VLAN filter # TODO: would be best if we did this directl via python, without calling # the iproute tooling try: subprocess.check_call(['ip', 'link', 'set', 'dev', pf, 'vf', vf_index, 'vlan', str(vlan_id)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except subprocess.CalledProcessError: raise RuntimeError( 'failed setting SR-IOV VLAN filter for vlan %s (ip link set command failed)' % vlan_name) def apply_sriov_config(config_manager, rootdir='/'): """ Go through all interfaces, identify which ones are SR-IOV VFs, create them and perform all other necessary setup. """ parser = libnetplan.Parser() parser.load_yaml_hierarchy(rootdir) np_state = libnetplan.State() np_state.import_parser_results(parser) config_manager.parse() interfaces = netifaces.interfaces() np_state = config_manager.np_state # for sr-iov devices, we identify VFs by them having a link: field # pointing to an PF. So let's browse through all ethernet devices, # find all that are VFs and count how many of those are linked to # particular PFs, as we need to then set the numvfs for each. vf_counts = defaultdict(int) # we also store all matches between VF/PF netplan entry names and # interface that they're currently matching to vfs = {} pfs = {} get_vf_count_and_functions( interfaces, np_state, vf_counts, vfs, pfs) # setup the required number of VFs per PF # at the same time store which PFs got changed in case the NICs # require some special quirks for the VF number to change vf_count_changed = [] if vf_counts: for pf, vf_count in vf_counts.items(): if not set_numvfs_for_pf(pf, vf_count): continue vf_count_changed.append(pf) if vf_count_changed: # some cards need special treatment when we want to change the # number of enabled VFs for pf in vf_count_changed: perform_hardware_specific_quirks(pf) # also, since the VF number changed, the interfaces list also # changed, so we need to refresh it interfaces = netifaces.interfaces() # now in theory we should have all the new VFs set up and existing; # this is needed because we will have to now match the defined VF # entries to existing interfaces, otherwise we won't be able to set # filtered VLANs for those. # XXX: does matching those even make sense? for vf in vfs: netdef = np_state[vf] if netdef.has_match: # right now we only match by name, as I don't think matching per # driver and/or macaddress makes sense # TODO: print warning if other matches are provided for interface in interfaces: if netdef.match_interface(itf_name=interface): if vf in vfs and vfs[vf]: raise ConfigurationError('matched more than one interface for a VF device: %s' % vf) vfs[vf] = interface else: if vf in interfaces: vfs[vf] = vf # Walk the SR-IOV PFs and check if we need to change the eswitch mode for netdef_id, iface in pfs.items(): netdef = np_state[netdef_id] eswitch_mode = netdef.embedded_switch_mode if eswitch_mode in ['switchdev', 'legacy']: pci_addr = _get_pci_slot_name(iface) pcidev = PCIDevice(pci_addr) if pcidev.is_pf: logging.debug("Found VFs of {}: {}".format(pcidev, pcidev.vf_addrs)) if pcidev.vfs: rebind_delayed = netdef.delay_virtual_functions_rebind try: unbind_vfs(pcidev.vfs, pcidev.driver) pcidev.devlink_set('eswitch', 'mode', eswitch_mode) finally: if not rebind_delayed: bind_vfs(pcidev.vfs, pcidev.driver) filtered_vlans_set = set() for vlan, netdef in np_state.vlans.items(): # there is a special sriov vlan renderer that one can use to mark # a selected vlan to be done in hardware (VLAN filtering) if netdef.has_sriov_vlan_filter: # this only works for SR-IOV VF interfaces link = netdef.vlan_link vlan_id = netdef.vlan_id vf = vfs.get(link.id) if not vf: # it is possible this is not an error, for instance when # the configuration has been defined 'for the future' # XXX: but maybe we should error out here as well? logging.warning( 'SR-IOV vlan defined for %s but link %s is either not a VF or has no matches' % (vlan, link.id)) continue # get the parent pf interface # first we fetch the related vf netplan entry # and finally, get the matched pf interface pf = pfs.get(link.sriov_link.id) if vf in filtered_vlans_set: raise ConfigurationError( 'interface %s for netplan device %s (%s) already has an SR-IOV vlan defined' % (vf, link.id, vlan)) # TODO: make sure that we don't apply the filter twice apply_vlan_filter_for_vf(pf, vf, vlan, vlan_id) filtered_vlans_set.add(vf) PK ! sx9Ň& �& utils.pynu �[��� # Copyright (C) 2018-2020 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # Author: Łukasz 'sil2100' Zemczak <lukasz.zemczak@canonical.com> # Author: Lukas 'slyon' Märdian <lukas.maerdian@canonical.com> # # 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; version 3. # # 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 logging import argparse import subprocess import netifaces import fnmatch import re import netplan.libnetplan as np from netplan.configmanager import ConfigurationError from netplan.libnetplan import LibNetplanException NM_SERVICE_NAME = 'NetworkManager.service' NM_SNAP_SERVICE_NAME = 'snap.network-manager.networkmanager.service' config_errors = (ConfigurationError, LibNetplanException, RuntimeError) def get_generator_path(): return os.environ.get('NETPLAN_GENERATE_PATH', '/usr/libexec/netplan/generate') def is_nm_snap_enabled(): return subprocess.call(['systemctl', '--quiet', 'is-enabled', NM_SNAP_SERVICE_NAME], stderr=subprocess.DEVNULL) == 0 def nmcli(args): # pragma: nocover (covered in autopkgtest) # 'nmcli' could be /usr/bin/nmcli or /snap/bin/nmcli -> /snap/bin/network-manager.nmcli # PATH is defined in cli/core.py subprocess.check_call(['nmcli'] + args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def nmcli_out(args: list) -> str: # pragma: nocover (covered in autopkgtest) # 'nmcli' could be /usr/bin/nmcli or /snap/bin/nmcli -> /snap/bin/network-manager.nmcli # PATH is defined in cli/core.py return subprocess.check_output(['nmcli'] + args, text=True) def nm_running(): # pragma: nocover (covered in autopkgtest) '''Check if NetworkManager is running''' try: nmcli(['general']) return True except (OSError, subprocess.SubprocessError): return False def nm_interfaces(paths, devices): pat = re.compile('^interface-name=(.*)$') interfaces = set() for path in paths: with open(path, 'r') as f: for line in f: m = pat.match(line) if m: # Expand/match globbing of interface names, to real devices interfaces.update(set(fnmatch.filter(devices, m.group(1)))) break # skip to next file return interfaces def systemctl_network_manager(action, sync=False): # If the network-manager snap is installed use its service # name rather than the one of the deb packaged NetworkManager if is_nm_snap_enabled(): return systemctl(action, [NM_SNAP_SERVICE_NAME], sync) return systemctl(action, [NM_SERVICE_NAME], sync) # pragma: nocover (covered in autopkgtest) def systemctl(action: str, services: list, sync: bool = False): if len(services) >= 1: command = ['systemctl', action] if not sync: command.append('--no-block') command.extend(services) subprocess.check_call(command) def networkd_interfaces(): interfaces = set() out = subprocess.check_output(['networkctl', '--no-pager', '--no-legend'], text=True) for line in out.splitlines(): s = line.strip().split(' ') if s[0].isnumeric() and s[-1] not in ['unmanaged', 'linger']: interfaces.add(s[0]) return interfaces def networkctl_reload(): subprocess.check_call(['networkctl', 'reload']) def networkctl_reconfigure(interfaces): if len(interfaces) >= 1: subprocess.check_call(['networkctl', 'reconfigure'] + list(interfaces)) def systemctl_is_active(unit_pattern): '''Return True if at least one matching unit is running''' if subprocess.call(['systemctl', '--quiet', 'is-active', unit_pattern]) == 0: return True return False def systemctl_is_masked(unit_pattern): '''Return True if output is "masked" or "masked-runtime"''' res = subprocess.run(['systemctl', 'is-enabled', unit_pattern], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) if res.returncode > 0 and 'masked' in res.stdout: return True return False def systemctl_daemon_reload(): '''Reload systemd unit files from disk and re-calculate its dependencies''' subprocess.check_call(['systemctl', 'daemon-reload']) def ip_addr_flush(iface): '''Flush all IP addresses of a given interface via iproute2''' subprocess.check_call(['ip', 'addr', 'flush', iface], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) def get_interface_driver_name(interface, only_down=False): # pragma: nocover (covered in autopkgtest) devdir = os.path.join('/sys/class/net', interface) if only_down: try: with open(os.path.join(devdir, 'operstate')) as f: state = f.read().strip() if state != 'down': logging.debug('device %s operstate is %s, not changing', interface, state) return None except IOError as e: logging.error('Cannot determine operstate of %s: %s', interface, str(e)) return None try: driver = os.path.realpath(os.path.join(devdir, 'device', 'driver')) driver_name = os.path.basename(driver) except IOError as e: logging.debug('Cannot replug %s: cannot read link %s/device: %s', interface, devdir, str(e)) return None return driver_name def get_interface_macaddress(interface): # return an empty list (and string) if no LL data can be found link = netifaces.ifaddresses(interface).get(netifaces.AF_LINK, [{}])[0] return link.get('addr', '') def find_matching_iface(interfaces: list, netdef): assert isinstance(netdef, np.NetDefinition) assert netdef.has_match matches = list(filter(lambda itf: netdef.match_interface( itf_name=itf, itf_driver=get_interface_driver_name(itf), itf_mac=get_interface_macaddress(itf)), interfaces)) # Return current name of unique matched interface, if available if len(matches) != 1: logging.info(matches) return None return matches[0] class NetplanCommand(argparse.Namespace): def __init__(self, command_id, description, leaf=True, testing=False): self.command_id = command_id self.description = description self.leaf_command = leaf self.testing = testing self._args = None self.debug = False self.breakpoint = False self.commandclass = None self.subcommands = {} self.subcommand = None self.func = None self.parser = argparse.ArgumentParser(prog="%s %s" % (sys.argv[0], command_id), description=description, add_help=True) self.parser.add_argument('--debug', action='store_true', help='Enable debug messages') self.parser.add_argument('--breakpoint', action='store_true', help=argparse.SUPPRESS) if not leaf: self.subparsers = self.parser.add_subparsers(title='Available commands', metavar='', dest='subcommand') p_help = self.subparsers.add_parser('help', description='Show this help message', help='Show this help message') p_help.set_defaults(func=self.print_usage) def update(self, args): self._args = args def parse_args(self): ns, self._args = self.parser.parse_known_args(args=self._args, namespace=self) if not self.subcommand and not self.leaf_command: print('You need to specify a command', file=sys.stderr) self.print_usage() def run_command(self): if self.commandclass: self.commandclass.update(self._args) # TODO: (cyphermox) this is actually testable in tests/cli.py; add it. if self.leaf_command and 'help' in self._args: # pragma: nocover (covered in autopkgtest) self.print_usage() if self.breakpoint: # pragma: nocover (cannot be automatically tested) breakpoint() self.func() def print_usage(self): self.parser.print_help(file=sys.stderr) sys.exit(os.EX_USAGE) def _add_subparser_from_class(self, name, commandclass): instance = commandclass() self.subcommands[name] = {} self.subcommands[name]['class'] = name self.subcommands[name]['instance'] = instance if instance.testing: if not os.environ.get('ENABLE_TEST_COMMANDS', None): return p = self.subparsers.add_parser(instance.command_id, description=instance.description, help=instance.description, add_help=False) p.set_defaults(func=instance.run, commandclass=instance) self.subcommands[name]['parser'] = p def _import_subcommands(self, submodules): import inspect for name, obj in inspect.getmembers(submodules): if inspect.isclass(obj) and issubclass(obj, NetplanCommand): self._add_subparser_from_class(name, obj) PK ! � � � __init__.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. PK ! �E�� � commands/info.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2019 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. '''netplan info command line''' import netplan.cli.utils as utils import netplan._features class NetplanInfo(utils.NetplanCommand): def __init__(self): super().__init__(command_id='info', description='Show available features', leaf=True) def run(self): # pragma: nocover (covered in autopkgtest) format_group = self.parser.add_mutually_exclusive_group(required=False) format_group.add_argument('--json', dest='version_format', action='store_const', const='json', help='Output version and features in JSON format') format_group.add_argument('--yaml', dest='version_format', action='store_const', const='yaml', help='Output version and features in YAML format') self.func = self.command_info self.parse_args() self.run_command() def command_info(self): netplan_version = { 'netplan.io': { 'website': 'https://netplan.io/', } } flags = netplan._features.NETPLAN_FEATURE_FLAGS netplan_version['netplan.io'].update({'features': flags}) # Default to output in YAML format. if self.version_format is None: self.version_format = 'yaml' if self.version_format == 'json': import json print(json.dumps(netplan_version, indent=2)) elif self.version_format == 'yaml': print('''netplan.io: website: "{}" features:'''.format(netplan_version['netplan.io']['website'])) for feature in netplan._features.NETPLAN_FEATURE_FLAGS: print(' - ' + feature) PK ! &�Fx x commands/set.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2020-2023 Canonical, Ltd. # Author: Lukas Märdian <slyon@ubuntu.com> # # 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; version 3. # # 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/>. '''netplan set command line''' import tempfile import re import io from netplan.cli.utils import NetplanCommand import netplan.libnetplan as libnetplan FALLBACK_FILENAME = '70-netplan-set.yaml' GLOBAL_KEYS = ['renderer', 'version'] class NetplanSet(NetplanCommand): def __init__(self): super().__init__(command_id='set', description='Add new setting by specifying a dotted key=value pair like ethernets.eth0.dhcp4=true', leaf=True) def run(self): self.parser.add_argument('key_value', type=str, help='The nested key=value pair in dotted format. Value can be NULL to delete a key.') self.parser.add_argument('--origin-hint', type=str, help='Can be used to help choose a name for the overwrite YAML file. \ A .yaml suffix will be appended automatically.') self.parser.add_argument('--root-dir', default='/', help='Overwrite configuration files in this root directory instead of /') self.func = self.command_set self.parse_args() self.run_command() def command_set(self): if self.origin_hint is not None and len(self.origin_hint) == 0: raise Exception('Invalid/empty origin-hint') if self.origin_hint: filename = '.'.join((self.origin_hint, 'yaml')) else: filename = None split = self.key_value.split('=', 1) if len(split) != 2: raise Exception('Invalid value specified') key, value = split if not key.startswith('network'): key = '.'.join(('network', key)) # Split the string into a list on the dot separators, and unescape the remaining dots yaml_path = [s.replace(r'\.', '.') for s in re.split(r'(?<!\\)\.', key)] parser = libnetplan.Parser() with tempfile.TemporaryFile() as tmp: libnetplan.create_yaml_patch(yaml_path, value, tmp) tmp.flush() # Load fields that are about to be deleted (e.g. some.setting=NULL) # Ignore those fields when parsing subsequent YAML files tmp.seek(0, io.SEEK_SET) parser.load_nullable_fields(tmp) # Parse the full, existing YAML config hierarchy parser.load_yaml_hierarchy(self.root_dir) # Load YAML patch, containing our update (new or deleted settings) tmp.seek(0, io.SEEK_SET) parser.load_yaml(tmp) # Validate the final parser state state = libnetplan.State() state.import_parser_results(parser) if filename: # only act on the output file (a.k.a. "origin-hint") parser_output_file = libnetplan.Parser() # Load fields that are about to be deleted ("some.setting=NULL") # Ignore those fields when parsing subsequent YAML files tmp.seek(0, io.SEEK_SET) parser_output_file.load_nullable_fields(tmp) # Load globals/netdefs that are to be ignored from the existing # YAML hierarchy, as our patch is supposed to override settings # in those netdefs via the output file. # Those netdefs and globals must end up in the output file # (a.k.a. "origin-hint", <filename>), have they been defined in # pre-existing YAML files or not. tmp.seek(0, io.SEEK_SET) parser_output_file.load_nullable_overrides(tmp, constraint=filename) # Parse the full YAML hierarchy and new patch, ignoring any # nullable overrides (netdefs/globals) from pre-existing files # and ignoring any nullable fields (settings to be deleted). # This way we can avoid updates to certain netdefs/globals to be # redirected into existing YAML files (defining those same # stanzas) or ignored, but have them written out to the single # output file. # XXX: The origin file of each individual YAML setting/stanza # should be tracked individually, to avoid this # double-parsing workaround (LP: #2003727) parser_output_file.load_yaml_hierarchy(self.root_dir) tmp.seek(0, io.SEEK_SET) parser_output_file.load_yaml(tmp) # Import the partial parser state, ignoring duplicated netdefs # from pre-existing YAML files, so we can force write the patch # contents to the output file or update this file if exists. state_output_file = libnetplan.State() state_output_file.import_parser_results(parser_output_file) state_output_file.write_yaml_file(filename, self.root_dir) else: state.update_yaml_hierarchy(FALLBACK_FILENAME, self.root_dir) PK ! $��)� � commands/generate.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. '''netplan generate command line''' import logging import os import sys import subprocess import shutil import netplan.cli.utils as utils class NetplanGenerate(utils.NetplanCommand): def __init__(self): super().__init__(command_id='generate', description='Generate backend specific configuration files' ' from /etc/netplan/*.yaml', leaf=True) def run(self): self.parser.add_argument('--root-dir', help='Search for and generate configuration files in this root directory instead of /') self.parser.add_argument('--mapping', help='Display the netplan device ID/backend/interface name mapping and exit.') self.func = self.command_generate self.parse_args() self.run_command() def command_generate(self): # if we are inside a snap, then call dbus to run netplan apply instead if "SNAP" in os.environ: # TODO: maybe check if we are inside a classic snap and don't do # this if we are in a classic snap? busctl = shutil.which("busctl") if busctl is None: raise RuntimeError("missing busctl utility") # pragma: nocover # XXX: DO NOT TOUCH or change this API call, it is used by snapd to communicate # using core20 netplan binary/client/CLI on core18 base systems. Any change # must be agreed upon with the snapd team, so we don't break support for # base systems running older netplan versions. # https://github.com/snapcore/snapd/pull/10212 res = subprocess.call([busctl, "call", "--quiet", "--system", "io.netplan.Netplan", # the service "/io/netplan/Netplan", # the object "io.netplan.Netplan", # the interface "Generate", # the method ]) if res != 0: if res == 130: raise PermissionError( "failed to communicate with dbus service") else: raise RuntimeError( "failed to communicate with dbus service: error %s" % res) else: return argv = [utils.get_generator_path()] if self.root_dir: argv += ['--root-dir', self.root_dir] if self.mapping: argv += ['--mapping', self.mapping] logging.debug('command generate: running %s', argv) # FIXME: os.execv(argv[0], argv) would be better but fails coverage sys.exit(subprocess.call(argv)) PK ! �P+'� � commands/try_command.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. '''netplan try command line''' import logging import os import time import shutil import signal import sys import tempfile from netplan.configmanager import ConfigManager import netplan.cli.utils as utils from netplan.cli.commands.apply import NetplanApply import netplan.terminal import netplan.libnetplan as libnetplan # Keep a timeout long enough to allow the network to converge, 60 seconds may # be slightly short given some complex configs, i.e. if STP must reconverge. DEFAULT_INPUT_TIMEOUT = 120 class NetplanTry(utils.NetplanCommand): def __init__(self): super().__init__(command_id='try', description='Try to apply a new netplan config to running ' 'system, with automatic rollback', leaf=True) self.configuration_changed = False self.new_interfaces = None self.config_file = None self._config_manager = None self.t_settings = None self.t = None self._rootdir = os.environ.get('DBUS_TEST_NETPLAN_ROOT', '/') self._netplan_try_stamp = os.path.join(self._rootdir, 'run', 'netplan', 'netplan-try.ready') @property def config_manager(self): # pragma: nocover (called by later commands) if not self._config_manager: self._config_manager = ConfigManager(prefix=self._rootdir) return self._config_manager def clear_ready_stamp(self): if os.path.isfile(self._netplan_try_stamp): os.remove(self._netplan_try_stamp) return True return False def touch_ready_stamp(self): os.makedirs(self._rootdir + '/run/netplan', mode=0o700, exist_ok=True) open(self._netplan_try_stamp, 'w').close() def run(self): # pragma: nocover (requires user input) self.parser.add_argument('--config-file', help='Apply the config file in argument in addition to current configuration.') self.parser.add_argument('--timeout', type=int, default=DEFAULT_INPUT_TIMEOUT, help="Maximum number of seconds to wait for the user's confirmation") self.parser.add_argument('--state', help='Directory containing previous YAML configuration') self.func = self.command_try self.parse_args() self.run_command() def command_try(self): # pragma: nocover (requires user input) if not self.is_revertable(): sys.exit(os.EX_CONFIG) try: fd = sys.stdin.fileno() self.t = netplan.terminal.Terminal(fd) self.t.save(self.t_settings) # we really don't want to be interrupted while doing backup/revert operations signal.signal(signal.SIGINT, self._signal_handler) signal.signal(signal.SIGUSR1, self._signal_handler) self.backup() self.setup() NetplanApply().command_apply(run_generate=True, sync=True, exit_on_error=False, state_dir=self.state) # Touch stamp file, it is the signal (for netplan-dbus) that we're # ready to accept any Accept/Reject input (like SIGUSR1 or SIGTERM) self.touch_ready_stamp() self.t.get_confirmation_input(timeout=self.timeout) except netplan.terminal.InputRejected: print("\nReverting.") self.revert() except netplan.terminal.InputAccepted: print("\nConfiguration accepted.") except Exception as e: print("\nAn error occurred: %s" % e) print("\nReverting.") self.revert() finally: if self.t: self.t.reset(self.t_settings) self.cleanup() self.clear_ready_stamp() def backup(self): # pragma: nocover (requires user input) backup_config_dir = False if self.config_file: backup_config_dir = True self.config_manager.backup(backup_config_dir=backup_config_dir) def setup(self): # pragma: nocover (requires user input) if self.config_file: dest_dir = os.path.join("/", "etc", "netplan") dest_name = os.path.basename(self.config_file).rstrip('.yaml') dest_suffix = time.time() dest_path = os.path.join(dest_dir, "{}.{}.yaml".format(dest_name, dest_suffix)) self.config_manager.add({self.config_file: dest_path}) self.configuration_changed = True def revert(self): # pragma: nocover (requires user input) # backup the state we just tried to apply tempdir = tempfile.mkdtemp() confdir = os.path.join(tempdir, 'etc', 'netplan') os.makedirs(confdir) shutil.copytree('/etc/netplan', confdir, dirs_exist_ok=True) # restore previous state self.config_manager.revert() NetplanApply().command_apply(run_generate=False, sync=True, exit_on_error=False, state_dir=tempdir) # clear the backup shutil.rmtree(tempdir) def cleanup(self): # pragma: nocover (requires user input) self.config_manager.cleanup() def is_revertable(self): ''' Check if the configuration is revertable, if it doesn't contain bits that we know are likely to render the system unstable if we apply it, or if we revert. Returns True if the parsed config is "revertable", meaning that we can actually rely on backends to re-apply /all/ of the relevant configuration to interfaces when their config changes. Returns False if the parsed config contains options that are known to not cleanly revert via the backend. ''' extra_config = [] if self.config_file: extra_config.append(self.config_file) np_state = None try: np_state = self.config_manager.parse(extra_config=extra_config) except utils.config_errors as e: logging.error(e) sys.exit(os.EX_CONFIG) revert_unsupported = [] # Bridges and bonds are special. They typically include (or could include) # more than one device in them, and they can be set with special parameters # to tweak their behavior, which are really hard to "revert", especially # as systemd-networkd doesn't necessarily touch them when config changes. multi_iface = {} # type: dict[str, libnetplan.NetDefinition] multi_iface.update(np_state.bridges) multi_iface.update(np_state.bonds) for itf in multi_iface.values(): if not itf.is_trivial_compound_itf: reason = "reverting custom parameters for bridges and bonds is not supported" revert_unsupported.append((itf.id, reason)) if revert_unsupported: for ifname, reason in revert_unsupported: print("{}: {}".format(ifname, reason)) print("\nPlease carefully review the configuration and use 'netplan apply' directly.") return False return True def _signal_handler(self, sig, frame): # pragma: nocover (requires user input) if sig == signal.SIGUSR1: raise netplan.terminal.InputAccepted() else: if self.configuration_changed: raise netplan.terminal.InputRejected() PK ! ��~��k �k commands/status.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2022 Canonical, Ltd. # Author: Lukas Märdian <slyon@ubuntu.com> # # 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; version 3. # # 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/>. '''netplan status command line''' import ipaddress import json import logging import re import socket import subprocess import sys from typing import Union, Dict, List, Type import dbus import yaml import netplan.cli.utils as utils JSON = Union[Dict[str, 'JSON'], List['JSON'], int, str, float, bool, Type[None]] MATCH_TAGS = re.compile(r'\[([a-z0-9]+)\].*\[\/\1\]') RICH_OUTPUT = False try: from rich.console import Console from rich.highlighter import RegexHighlighter from rich.theme import Theme class NetplanHighlighter(RegexHighlighter): base_style = 'netplan.' highlights = [ r'(^|[\s\/])(?P<int>\d+)([\s:]?\s|$)', r'(?P<str>(\"|\').+(\"|\'))', ] RICH_OUTPUT = True except ImportError: # pragma: nocover (we mock RICH_OUTPUT, ignore the logging) logging.debug("python3-rich not found, falling back to plain output") class Interface(): def __extract_mac(self, ip: dict) -> str: ''' Extract the MAC address if it's set inside the JSON data and seems to have the correct format. Return 'None' otherwise. ''' if len(address := ip.get('address', '')) == 17: # 6 byte MAC (+5 colons) return address.lower() return None def __init__(self, ip: dict, nd_data: JSON = [], nm_data: JSON = [], resolved_data: tuple = (None, None), route_data: tuple = (None, None)): self.idx: int = ip.get('ifindex', -1) self.name: str = ip.get('ifname', 'unknown') self.adminstate: str = 'UP' if 'UP' in ip.get('flags', []) else 'DOWN' self.operstate: str = ip.get('operstate', 'unknown').upper() self.macaddress: str = self.__extract_mac(ip) # Filter networkd/NetworkManager data nm_data = nm_data or [] # avoid 'None' value on systems without NM self.nd: JSON = next((x for x in nd_data if x['Index'] == self.idx), None) self.nm: JSON = next((x for x in nm_data if x['device'] == self.name), None) # Filter resolved's DNS data self.dns_addresses: list = None if resolved_data[0]: self.dns_addresses = [] for itr in resolved_data[0]: if int(itr[0]) == int(self.idx): ipfamily = itr[1] dns = itr[2] self.dns_addresses.append(socket.inet_ntop(ipfamily, b''.join([v.to_bytes(1, 'big') for v in dns]))) self.dns_search: list = None if resolved_data[1]: self.dns_search = [] for v in resolved_data[1]: if int(v[0]) == int(self.idx): self.dns_search.append(str(v[1])) # Filter route data _routes: list = [] self.routes: list = None if route_data[0]: _routes += route_data[0] if route_data[1]: _routes += route_data[1] if _routes: self.routes = [] for obj in _routes: if obj.get('dev') == self.name: elem = {'to': obj.get('dst')} val = obj.get('gateway') if val: elem['via'] = val val = obj.get('prefsrc') if val: elem['from'] = val val = obj.get('metric') if val: elem['metric'] = val val = obj.get('type') if val: elem['type'] = val val = obj.get('scope') if val: elem['scope'] = val val = obj.get('protocol') if val: elem['protocol'] = val self.routes.append(elem) self.addresses: list = None if addr_info := ip.get('addr_info'): self.addresses = [] for addr in addr_info: flags: list = [] if ipaddress.ip_address(addr['local']).is_link_local: flags.append('link') if self.routes: for route in self.routes: if ('from' in route and ipaddress.ip_address(route['from']) == ipaddress.ip_address(addr['local'])): if route['protocol'] == 'dhcp': flags.append('dhcp') break ip_addr = addr['local'].lower() elem = {ip_addr: {'prefix': addr['prefixlen']}} if flags: elem[ip_addr]['flags'] = flags self.addresses.append(elem) self.iproute_type: str = None if info_kind := ip.get('linkinfo', {}).get('info_kind'): self.iproute_type = info_kind.strip() # workaround: query some data which is not available via networkctl's JSON output self._networkctl: str = self.query_networkctl(self.name) or '' def query_nm_ssid(self, con_name: str) -> str: ssid: str = None try: ssid = utils.nmcli_out(['--get-values', '802-11-wireless.ssid', 'con', 'show', 'id', con_name]) return ssid.strip() except Exception as e: logging.warning('Cannot query NetworkManager SSID for {}: {}'.format( con_name, str(e))) return ssid def query_networkctl(self, ifname: str) -> str: output: str = None try: output = subprocess.check_output(['networkctl', 'status', '--', ifname], text=True) except Exception as e: logging.warning('Cannot query networkctl for {}: {}'.format( ifname, str(e))) return output def json(self) -> JSON: json = { 'index': self.idx, 'adminstate': self.adminstate, 'operstate': self.operstate, } if self.type: json['type'] = self.type if self.ssid: json['ssid'] = self.ssid if self.tunnel_mode: json['tunnel_mode'] = self.tunnel_mode if self.backend: json['backend'] = self.backend if self.netdef_id: json['id'] = self.netdef_id if self.macaddress: json['macaddress'] = self.macaddress if self.vendor: json['vendor'] = self.vendor if self.addresses: json['addresses'] = self.addresses if self.dns_addresses: json['dns_addresses'] = self.dns_addresses if self.dns_search: json['dns_search'] = self.dns_search if self.routes: json['routes'] = self.routes if self.activation_mode: json['activation_mode'] = self.activation_mode return (self.name, json) @property def up(self) -> bool: return self.adminstate == 'UP' and self.operstate == 'UP' @property def down(self) -> bool: return self.adminstate == 'DOWN' and self.operstate == 'DOWN' @property def type(self) -> str: match = dict({ 'bond': 'bond', 'bridge': 'bridge', 'ether': 'ethernet', 'ipgre': 'tunnel', 'ip6gre': 'tunnel', 'loopback': 'ethernet', 'sit': 'tunnel', 'tunnel': 'tunnel', 'tunnel6': 'tunnel', 'wireguard': 'tunnel', 'wlan': 'wifi', 'wwan': 'modem', 'vrf': 'vrf', 'vxlan': 'tunnel', }) nd_type = self.nd.get('Type') if self.nd else None if nd_type in match: return match[nd_type] logging.warning('Unknown device type: {}'.format(nd_type)) return None @property def tunnel_mode(self) -> str: if self.type == 'tunnel' and self.iproute_type: return self.iproute_type return None @property def backend(self) -> str: if (self.nd and 'unmanaged' not in self.nd.get('SetupState', '') and 'run/systemd/network/10-netplan-' in self.nd.get('NetworkFile', '')): return 'networkd' elif self.nm and 'run/NetworkManager/system-connections/netplan-' in self.nm.get('filename', ''): return 'NetworkManager' return None @property def netdef_id(self) -> str: if self.backend == 'networkd': return self.nd.get('NetworkFile', '').split( 'run/systemd/network/10-netplan-')[1].split('.network')[0] elif self.backend == 'NetworkManager': netdef = self.nm.get('filename', '').split( 'run/NetworkManager/system-connections/netplan-')[1].split('.nmconnection')[0] if self.nm.get('type', '') == '802-11-wireless': ssid = self.query_nm_ssid(self.nm.get('name')) if ssid: # XXX: escaping needed? netdef = netdef.split('-' + ssid)[0] return netdef return None @property def vendor(self) -> str: if self.nd and 'Vendor' in self.nd and self.nd['Vendor']: return self.nd['Vendor'].strip() return None @property def ssid(self) -> str: if self.type == 'wifi': # XXX: available from networkctl's JSON output as of v250: # https://github.com/systemd/systemd/commit/da7c995 for line in self._networkctl.splitlines(): line = line.strip() key = 'WiFi access point: ' if line.startswith(key): ssid = line[len(key):-len(' (xB:SS:ID:xx:xx:xx)')].strip() return ssid if ssid else None return None @property def activation_mode(self) -> str: if self.backend == 'networkd': # XXX: available from networkctl's JSON output as of v250: # https://github.com/systemd/systemd/commit/3b60ede for line in self._networkctl.splitlines(): line = line.strip() key = 'Activation Policy: ' if line.startswith(key): mode = line[len(key):].strip() return mode if mode != 'up' else None # XXX: this is not fully supported on NetworkManager, only 'manual'/'up' elif self.backend == 'NetworkManager': return 'manual' if self.nm['autoconnect'] == 'no' else None return None class NetplanStatus(utils.NetplanCommand): def __init__(self): super().__init__(command_id='status', description='Query networking state of the running system', leaf=True) self.all = False def run(self): self.parser.add_argument('ifname', nargs='?', type=str, default=None, help='Show only this interface') self.parser.add_argument('-a', '--all', action='store_true', help='Show all interface data (incl. inactive)') self.parser.add_argument('-f', '--format', default='tabular', help='Output in machine readable `json` or `yaml` format') self.func = self.command self.parse_args() self.run_command() def resolvconf_json(self) -> dict: res = { 'addresses': [], 'search': [], 'mode': None, } try: with open('/etc/resolv.conf') as f: # check first line for systemd-resolved stub or compat modes firstline = f.readline() if '# This is /run/systemd/resolve/stub-resolv.conf' in firstline: res['mode'] = 'stub' elif '# This is /run/systemd/resolve/resolv.conf' in firstline: res['mode'] = 'compat' for line in [firstline] + f.readlines(): if line.startswith('nameserver'): res['addresses'] += line.split()[1:] # append if line.startswith('search'): res['search'] = line.split()[1:] # override except Exception as e: logging.warning('Cannot parse /etc/resolv.conf: {}'.format(str(e))) return res def query_online_state(self, interfaces: list) -> bool: # TODO: fully implement network-online.target specification (FO020): # https://discourse.ubuntu.com/t/spec-definition-of-an-online-system/27838 for itf in interfaces: if itf.up and itf.addresses and itf.routes and itf.dns_addresses: non_local_ips = [] for addr in itf.addresses: ip, extra = list(addr.items())[0] if 'flags' not in extra or 'link' not in extra['flags']: non_local_ips.append(ip) default_routes = [x for x in itf.routes if x.get('to', None) == 'default'] if non_local_ips and default_routes and itf.dns_addresses: return True return False def process_generic(self, cmd_output: str) -> JSON: return json.loads(cmd_output) def query_iproute2(self) -> JSON: data: JSON = None try: output: str = subprocess.check_output(['ip', '-d', '-j', 'addr'], text=True) data = self.process_generic(output) except Exception as e: logging.critical('Cannot query iproute2 interface data: {}'.format(str(e))) return data def process_networkd(self, cmd_output) -> JSON: return json.loads(cmd_output)['Interfaces'] def query_networkd(self) -> JSON: data: JSON = None try: output: str = subprocess.check_output(['networkctl', '--json=short'], text=True) data = self.process_networkd(output) except Exception as e: logging.critical('Cannot query networkd interface data: {}'.format(str(e))) return data def process_nm(self, cmd_output) -> JSON: data: JSON = [] for line in cmd_output.splitlines(): split = line.split(':') dev = split[0] if split[0] else None if dev: # ignore inactive connection profiles data.append({ 'device': dev, 'name': split[1], 'uuid': split[2], 'filename': split[3], 'type': split[4], 'autoconnect': split[5], }) return data def query_nm(self) -> JSON: data: JSON = None try: output: str = utils.nmcli_out(['-t', '-f', 'DEVICE,NAME,UUID,FILENAME,TYPE,AUTOCONNECT', 'con', 'show']) data = self.process_nm(output) except Exception as e: logging.debug('Cannot query NetworkManager interface data: {}'.format(str(e))) return data def query_routes(self) -> tuple: data4 = None data6 = None try: output4: str = subprocess.check_output(['ip', '-d', '-j', 'route'], text=True) data4: JSON = self.process_generic(output4) output6: str = subprocess.check_output(['ip', '-d', '-j', '-6', 'route'], text=True) data6: JSON = self.process_generic(output6) except Exception as e: logging.debug('Cannot query iproute2 route data: {}'.format(str(e))) return (data4, data6) def query_resolved(self) -> tuple: addresses = None search = None try: ipc = dbus.SystemBus() resolve1 = ipc.get_object('org.freedesktop.resolve1', '/org/freedesktop/resolve1') resolve1_if = dbus.Interface(resolve1, 'org.freedesktop.DBus.Properties') res = resolve1_if.GetAll('org.freedesktop.resolve1.Manager') addresses = res['DNS'] search = res['Domains'] except Exception as e: logging.debug('Cannot query resolved DNS data: {}'.format(str(e))) return (addresses, search) def plain_print(self, *args, **kwargs): if len(args): lst = list(args) for tag in MATCH_TAGS.findall(lst[0]): # remove matching opening and closing tag lst[0] = lst[0].replace('[{}]'.format(tag), '')\ .replace('[/{}]'.format(tag), '') return print(*lst, **kwargs) return print(*args, **kwargs) def pretty_print(self, data: JSON, total: int, _console_width=None) -> None: if RICH_OUTPUT: # TODO: Use a proper (subiquity?) color palette theme = Theme({ 'netplan.int': 'bold cyan', 'netplan.str': 'yellow', 'muted': 'grey62', 'online': 'green bold', 'offline': 'red bold', 'unknown': 'yellow bold', 'highlight': 'bold' }) console = Console(highlighter=NetplanHighlighter(), theme=theme, width=_console_width, emoji=False) pprint = console.print else: pprint = self.plain_print pad = '18' global_state = data.get('netplan-global-state', {}) interfaces = [(key, data[key]) for key in data if key != 'netplan-global-state'] # Global state pprint(('{title:>'+pad+'} {value}').format( title='Online state:', value='[online]online[/online]' if global_state.get('online', False) else '[offline]offline[/offline]', )) ns = global_state.get('nameservers', {}) dns_addr: list = ns.get('addresses', []) dns_mode: str = ns.get('mode') dns_search: list = ns.get('search', []) if dns_addr: for i, val in enumerate(dns_addr): pprint(('{title:>'+pad+'} {value}[muted]{mode}[/muted]').format( title='DNS Addresses:' if i == 0 else '', value=val, mode=' ({})'.format(dns_mode) if dns_mode else '', )) if dns_search: for i, val in enumerate(dns_search): pprint(('{title:>'+pad+'} {value}').format( title='DNS Search:' if i == 0 else '', value=val, )) pprint() # Per interface for (ifname, data) in interfaces: state = data.get('operstate', 'UNKNOWN') + '/' + data.get('adminstate', 'UNKNOWN') scolor = 'unknown' if state == 'UP/UP': state = 'UP' scolor = 'online' elif state == 'DOWN/DOWN': state = 'DOWN' scolor = 'offline' full_type = data.get('type', 'other') ssid = data.get('ssid') tunnel_mode = data.get('tunnel_mode') if full_type == 'wifi' and ssid: full_type += ('/"' + ssid + '"') elif full_type == 'tunnel' and tunnel_mode: full_type += ('/' + tunnel_mode) pprint('[{col}]●[/{col}] {idx:>2}: {name} {type} [{col}]{state}[/{col}] ({backend}{netdef})'.format( col=scolor, idx=data.get('index', '?'), name=ifname, type=full_type, state=state, backend=data.get('backend', 'unmanaged'), netdef=': [highlight]{}[/highlight]'.format(data.get('id')) if data.get('id') else '' )) if data.get('macaddress'): pprint(('{title:>'+pad+'} {mac}[muted]{vendor}[/muted]').format( title='MAC Address:', mac=data.get('macaddress', ''), vendor=' ({})'.format(data.get('vendor', '')) if data.get('vendor') else '', )) lst: list = data.get('addresses', []) if lst: for i, obj in enumerate(lst): ip, extra = list(obj.items())[0] # get first (any only) address flags = [] if extra.get('flags'): # flags flags = extra.get('flags', []) highlight_start = '' highlight_end = '' if not flags or 'dhcp' in flags: highlight_start = '[highlight]' highlight_end = '[/highlight]' pprint(('{title:>'+pad+'} {start}{ip}/{prefix}{end}[muted]{extra}[/muted]').format( title='Addresses:' if i == 0 else '', ip=ip, prefix=extra.get('prefix', ''), extra=' ('+', '.join(flags)+')' if flags else '', start=highlight_start, end=highlight_end, )) lst = data.get('dns_addresses', []) if lst: for i, val in enumerate(lst): pprint(('{title:>'+pad+'} {value}').format( title='DNS Addresses:' if i == 0 else '', value=val, )) lst = data.get('dns_search', []) if lst: for i, val in enumerate(lst): pprint(('{title:>'+pad+'} {value}').format( title='DNS Search:' if i == 0 else '', value=val, )) lst = data.get('routes', []) if lst: for i, obj in enumerate(lst): default_start = '' default_end = '' if obj['to'] == 'default': default_start = '[highlight]' default_end = '[/highlight]' via = '' if 'via' in obj: via = ' via ' + obj['via'] src = '' if 'from' in obj: src = ' from ' + obj['from'] metric = '' if 'metric' in obj: metric = ' metric ' + str(obj['metric']) extra = [] if 'protocol' in obj and obj['protocol'] != 'kernel': proto = obj['protocol'] extra.append(proto) if 'scope' in obj and obj['scope'] != 'global': scope = obj['scope'] extra.append(scope) if 'type' in obj and obj['type'] != 'unicast': type = obj['type'] extra.append(type) pprint(('{title:>'+pad+'} {start}{to}{via}{src}{metric}{end}[muted]{extra}[/muted]').format( title='Routes:' if i == 0 else '', to=obj['to'], via=via, src=src, metric=metric, extra=' ('+', '.join(extra)+')' if extra else '', start=default_start, end=default_end)) val = data.get('activation_mode') if val: pprint(('{title:>'+pad+'} {value}').format( title='Activation Mode:', value=val, )) pprint() hidden = total - len(interfaces) if (hidden > 0): pprint('{} inactive interfaces hidden. Use "--all" to show all.'.format(hidden)) def command(self): # Make sure sd-networkd is running, as we need the data it provides. if not utils.systemctl_is_active('systemd-networkd.service'): if utils.systemctl_is_masked('systemd-networkd.service'): logging.error('\'netplan status\' depends on networkd, ' 'but systemd-networkd.service is masked. ' 'Please start it.') sys.exit(1) logging.debug('systemd-networkd.service is not active. Starting...') utils.systemctl('start', ['systemd-networkd.service'], True) # required data: iproute2 and sd-networkd can be expected to exist, # due to hard package dependencies iproute2 = self.query_iproute2() networkd = self.query_networkd() if not iproute2 or not networkd: logging.error('Could not query iproute2 or systemd-networkd') sys.exit(1) # optional data nmcli = self.query_nm() route4, route6 = self.query_routes() dns_addresses, dns_search = self.query_resolved() interfaces = [Interface(itf, networkd, nmcli, (dns_addresses, dns_search), (route4, route6)) for itf in iproute2] total = len(interfaces) # show only active interfaces by default filtered = [itf for itf in interfaces if itf.operstate != 'DOWN'] # down interfaces do not contribute anything to the online state online_state = self.query_online_state(filtered) # show only a single interface, if requested # XXX: bash completion (for interfaces names) if self.ifname: filtered = [next((itf for itf in interfaces if itf.name == self.ifname), None)] filtered = [elem for elem in filtered if elem is not None] if self.ifname and filtered == []: logging.error('Could not find interface {}'.format(self.ifname)) sys.exit(1) # Global state data = { 'netplan-global-state': { 'online': online_state, 'nameservers': self.resolvconf_json() } } # Per interface itf_iter = interfaces if self.all else filtered for itf in itf_iter: ifname, obj = itf.json() data[ifname] = obj # Output data in requested format output_format = self.format.lower() if output_format == 'json': # structural JSON output print(json.dumps(data, indent=None)) elif output_format == 'yaml': # stuctural YAML output print(yaml.dump(data, default_flow_style=False)) else: # pretty print, human readable output self.pretty_print(data, total) PK ! I��O �O commands/migrate.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. '''netplan migrate command line''' import logging import os import sys import re from glob import glob try: import yaml NO_YAML = False except ImportError: # pragma: nocover NO_YAML = True from collections import OrderedDict import ipaddress import netplan.cli.utils as utils class NetplanMigrate(utils.NetplanCommand): def __init__(self): super().__init__(command_id='migrate', description='Migration of /etc/network/interfaces to netplan', leaf=True, testing=True) def parse_dns_options(self, if_options, if_config): """Parse dns options (dns-nameservers and dns-search) from if_options (an interface options dict) into the interface configuration if_config Mutates the arguments in place. """ if 'dns-nameservers' in if_options: if 'nameservers' not in if_config: if_config['nameservers'] = {} if 'addresses' not in if_config['nameservers']: if_config['nameservers']['addresses'] = [] for ns in if_options['dns-nameservers'].split(' '): # allow multiple spaces in the dns-nameservers entry if not ns: continue # validate? if_config['nameservers']['addresses'] += [ns] del if_options['dns-nameservers'] if 'dns-search' in if_options: if 'nameservers' not in if_config: if_config['nameservers'] = {} if 'search' not in if_config['nameservers']: if_config['nameservers']['search'] = [] for domain in if_options['dns-search'].split(' '): # allow multiple spaces in the dns-search entry if not domain: continue if_config['nameservers']['search'] += [domain] del if_options['dns-search'] def parse_mtu(self, iface, if_options, if_config): """Parse out the MTU. Operates the same way as parse_dns_options iface is the name of the interface, used only to print error messages """ if 'mtu' in if_options: try: mtu = int(if_options['mtu']) except ValueError: logging.error('%s: cannot parse "%s" as an MTU', iface, if_options['mtu']) sys.exit(2) if 'mtu' in if_config and not if_config['mtu'] == mtu: logging.error('%s: tried to set MTU=%d, but already have MTU=%d', iface, mtu, if_config['mtu']) sys.exit(2) if_config['mtu'] = mtu del if_options['mtu'] def parse_hwaddress(self, iface, if_options, if_config): """Parse out the manually configured MAC. Operates the same way as parse_dns_options iface is the name of the interface, used only to print error messages """ if 'hwaddress' in if_options: if 'macaddress' in if_config and not if_config['macaddress'] == if_options['hwaddress']: logging.error('%s: tried to set MAC %s, but already have MAC %s', iface, if_options['hwaddress'], if_config['macaddress']) sys.exit(2) if_config['macaddress'] = if_options['hwaddress'] del if_options['hwaddress'] def run(self): self.parser.add_argument('--root-dir', help='Search for and generate configuration files in this root directory instead of /') self.parser.add_argument('--dry-run', action='store_true', help='Print converted netplan configuration to stdout instead of writing/changing files') self.func = self.command_migrate self.parse_args() if NO_YAML: # pragma: nocover logging.error("""The `yaml` Python package couldn't be imported, and is needed for the migrate command. To install it on Debian or Ubuntu-based system, run `apt install python3-yaml`""") sys.exit(1) self.run_command() def command_migrate(self): netplan_config = {} try: ifaces, auto_ifaces = self.parse_ifupdown(self.root_dir or '') except ValueError as e: logging.error(str(e)) sys.exit(2) for iface, family_config in ifaces.items(): for family, config in family_config.items(): logging.debug('Converting %s family %s %s', iface, family, config) if iface not in auto_ifaces: logging.error('%s: non-automatic interfaces are not supported', iface) sys.exit(2) if config['method'] == 'loopback': # both systemd and modern ifupdown set up lo automatically logging.debug('Ignoring loopback interface %s', iface) elif config['method'] == 'dhcp': c = netplan_config.setdefault('network', {}).setdefault('ethernets', {}).setdefault(iface, {}) self.parse_dns_options(config['options'], c) self.parse_hwaddress(iface, config['options'], c) if config['options']: logging.error('%s: option(s) %s are not supported for dhcp method', iface, ", ".join(config['options'].keys())) sys.exit(2) if family == 'inet': c['dhcp4'] = True else: assert family == 'inet6' c['dhcp6'] = True elif config['method'] == 'static': c = netplan_config.setdefault('network', {}).setdefault('ethernets', {}).setdefault(iface, {}) if 'addresses' not in c: c['addresses'] = [] self.parse_dns_options(config['options'], c) self.parse_mtu(iface, config['options'], c) self.parse_hwaddress(iface, config['options'], c) # ipv4 if family == 'inet': # Already handled: mtu, hwaddress # Supported: address netmask gateway # Not supported yet: metric(?) # No YAML support: pointopoint scope broadcast supported_opts = set(['address', 'netmask', 'gateway']) unsupported_opts = set(['broadcast', 'metric', 'pointopoint', 'scope']) opts = set(config['options'].keys()) bad_opts = opts - supported_opts if bad_opts: for unsupported in bad_opts.intersection(unsupported_opts): logging.error('%s: unsupported %s option "%s"', iface, family, unsupported) sys.exit(2) for unknown in bad_opts - unsupported_opts: logging.error('%s: unknown %s option "%s"', iface, family, unknown) sys.exit(2) # the address may contain a /prefix suffix, or # the netmask property may be used. It's not clear # what happens if both are supplied. if 'address' not in config['options']: logging.error('%s: no address supplied in static method', iface) sys.exit(2) if '/' in config['options']['address']: addr_spec = config['options']['address'].split('/')[0] net_spec = config['options']['address'] else: if 'netmask' not in config['options']: logging.error('%s: address does not specify prefix length, and netmask not specified', iface) sys.exit(2) addr_spec = config['options']['address'] net_spec = config['options']['address'] + '/' + config['options']['netmask'] try: ipaddr = ipaddress.IPv4Address(addr_spec) except ipaddress.AddressValueError as a: logging.error('%s: error parsing "%s" as an IPv4 address: %s', iface, addr_spec, a) sys.exit(2) try: ipnet = ipaddress.IPv4Network(net_spec, strict=False) except ipaddress.NetmaskValueError as a: logging.error('%s: error parsing "%s" as an IPv4 network: %s', iface, net_spec, a) sys.exit(2) c['addresses'] += [str(ipaddr) + '/' + str(ipnet.prefixlen)] if 'gateway' in config['options']: # validate? c['gateway4'] = config['options']['gateway'] # ipv6 else: assert family == 'inet6' # Already handled: mtu, hwaddress # supported: address netmask gateway # partially supported: accept_ra (0/1 supported, 2 has no YAML rep) # unsupported: metric(?) # no YAML representation: media autoconf privext scope # preferred-lifetime dad-attempts dad-interval supported_opts = set(['address', 'netmask', 'gateway', 'accept_ra']) unsupported_opts = set(['metric', 'media', 'autoconf', 'privext', 'scope', 'preferred-lifetime', 'dad-attempts', 'dad-interval']) opts = set(config['options'].keys()) bad_opts = opts - supported_opts if bad_opts: for unsupported in bad_opts.intersection(unsupported_opts): logging.error('%s: unsupported %s option "%s"', iface, family, unsupported) sys.exit(2) for unknown in bad_opts - unsupported_opts: logging.error('%s: unknown %s option "%s"', iface, family, unknown) sys.exit(2) # the address may contain a /prefix suffix, or # the netmask property may be used. It's not clear # what happens if both are supplied. if 'address' not in config['options']: logging.error('%s: no address supplied in static method', iface) sys.exit(2) if '/' in config['options']['address']: addr_spec = config['options']['address'].split('/')[0] net_spec = config['options']['address'] else: if 'netmask' not in config['options']: logging.error('%s: address does not specify prefix length, and netmask not specified', iface) sys.exit(2) addr_spec = config['options']['address'] net_spec = config['options']['address'] + '/' + config['options']['netmask'] try: ipaddr = ipaddress.IPv6Address(addr_spec) except ipaddress.AddressValueError as a: logging.error('%s: error parsing "%s" as an IPv6 address: %s', iface, addr_spec, a) sys.exit(2) try: ipnet = ipaddress.IPv6Network(net_spec, strict=False) except ipaddress.NetmaskValueError as a: logging.error('%s: error parsing "%s" as an IPv6 network: %s', iface, net_spec, a) sys.exit(2) c['addresses'] += [str(ipaddr) + '/' + str(ipnet.prefixlen)] if 'gateway' in config['options']: # validate? c['gateway6'] = config['options']['gateway'] if 'accept_ra' in config['options']: if config['options']['accept_ra'] == '0': c['accept_ra'] = False elif config['options']['accept_ra'] == '1': c['accept_ra'] = True elif config['options']['accept_ra'] == '2': logging.error('%s: netplan does not support accept_ra=2', iface) sys.exit(2) else: logging.error('%s: unexpected accept_ra value "%s"', iface, config['options']['accept_ra']) sys.exit(2) else: # pragma nocover # this should be unreachable logging.error('%s: method %s is not supported', iface, config['method']) sys.exit(2) if_config = os.path.join(self.root_dir or '/', 'etc/network/interfaces') if netplan_config: netplan_config['network']['version'] = 2 netplan_yaml = yaml.dump(netplan_config) if self.dry_run: print(netplan_yaml) else: dest = os.path.join(self.root_dir or '/', 'etc/netplan/10-ifupdown.yaml') try: os.makedirs(os.path.dirname(dest)) except FileExistsError: pass try: with open(dest, 'x') as f: f.write(netplan_yaml) except FileExistsError: logging.error('%s already exists; remove it if you want to run the migration again', dest) sys.exit(3) logging.info('migration complete, wrote %s', dest) else: logging.warning('ifupdown does not configure any interfaces, nothing to migrate') if not self.dry_run: logging.info('renaming %s to %s.netplan-converted', if_config, if_config) os.rename(if_config, if_config + '.netplan-converted') def _ifupdown_lines_from_file(self, rootdir, path): '''Return normalized lines from ifupdown config This resolves "source" and "source-directory" includes. ''' def expand_source_arg(rootdir, curdir, line): arg = line.split()[1] if arg.startswith('/'): return rootdir + arg else: return curdir + '/' + arg lines = [] rootdir_len = len(rootdir) + 1 try: with open(rootdir + '/' + path) as f: logging.debug('reading %s', f.name) for line in f: # normalize, strip empty lines and comments line = line.strip() if not line or line.startswith('#'): continue if line.startswith('source-directory '): valid_re = re.compile('^[a-zA-Z0-9_-]+$') d = expand_source_arg(rootdir, os.path.dirname(f.name), line) for f in os.listdir(d): if valid_re.match(f): lines += self._ifupdown_lines_from_file(rootdir, os.path.join(d[rootdir_len:], f)) elif line.startswith('source '): for f in glob(expand_source_arg(rootdir, os.path.dirname(f.name), line)): lines += self._ifupdown_lines_from_file(rootdir, f[rootdir_len:]) else: lines.append(line) except FileNotFoundError: logging.debug('%s/%s does not exist, ignoring', rootdir, path) return lines def parse_ifupdown(self, rootdir='/'): '''Parse ifupdown configuration. Return (iface_name → family → {method, options}, auto_ifaces: set) tuple on successful parsing, or a ValueError when encountering an invalid file or ifupdown features which are not supported (such as "mapping"). options is itself a dictionary option_name → value. ''' # expected number of fields for every possible keyword, excluding the keyword itself fieldlen = {'auto': 1, 'allow-auto': 1, 'allow-hotplug': 1, 'mapping': 1, 'no-scripts': 1, 'iface': 3} # read and normalize all lines from config, with resolving includes lines = self._ifupdown_lines_from_file(rootdir, '/etc/network/interfaces') ifaces = OrderedDict() auto = set() in_options = None # interface name if parsing options lines after iface stanza in_family = None # we now have resolved all includes and normalized lines for line in lines: fields = line.split() try: # does the line start with a known stanza field? exp_len = fieldlen[fields[0]] logging.debug('line fields %s (expected length: %i)', fields, exp_len) in_options = None # stop option line parsing of iface stanza in_family = None except KeyError: # no known stanza field, are we in an iface stanza and parsing options? if in_options: logging.debug('in_options %s, parsing as option: %s', in_options, line) ifaces[in_options][in_family]['options'][fields[0]] = line.split(maxsplit=1)[1] continue else: raise ValueError('Unknown stanza type %s' % fields[0]) # do we have the expected #parameters? if len(fields) != exp_len + 1: raise ValueError('Expected %i fields for stanza type %s but got %i' % (exp_len, fields[0], len(fields) - 1)) # we have a valid stanza line now, handle them if fields[0] in ('auto', 'allow-auto', 'allow-hotplug'): auto.add(fields[1]) elif fields[0] == 'mapping': raise ValueError('mapping stanza is not supported') elif fields[0] == 'no-scripts': pass # ignore these elif fields[0] == 'iface': if fields[2] not in ('inet', 'inet6'): raise ValueError('Unknown address family %s' % fields[2]) if fields[3] not in ('loopback', 'static', 'dhcp'): raise ValueError('Unsupported method %s' % fields[3]) in_options = fields[1] in_family = fields[2] ifaces.setdefault(fields[1], OrderedDict())[in_family] = {'method': fields[3], 'options': {}} else: raise NotImplementedError('stanza type %s is not implemented' % fields[0]) # pragma nocover logging.debug('final parsed interfaces: %s; auto ifaces: %s', ifaces, auto) return (ifaces, auto) PK ! Fﱗ� � commands/__init__.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. from netplan.cli.commands.apply import NetplanApply from netplan.cli.commands.generate import NetplanGenerate from netplan.cli.commands.ip import NetplanIp from netplan.cli.commands.migrate import NetplanMigrate from netplan.cli.commands.try_command import NetplanTry from netplan.cli.commands.info import NetplanInfo from netplan.cli.commands.set import NetplanSet from netplan.cli.commands.get import NetplanGet from netplan.cli.commands.sriov_rebind import NetplanSriovRebind from netplan.cli.commands.status import NetplanStatus __all__ = [ 'NetplanApply', 'NetplanGenerate', 'NetplanIp', 'NetplanMigrate', 'NetplanTry', 'NetplanInfo', 'NetplanSet', 'NetplanGet', 'NetplanSriovRebind', 'NetplanStatus', ] PK ! �$<�O �O commands/apply.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018-2020 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # Author: Łukasz 'sil2100' Zemczak <lukasz.zemczak@canonical.com> # Author: Lukas 'slyon' Märdian <lukas.maerdian@canonical.com> # # 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; version 3. # # 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/>. '''netplan apply command line''' import logging import os import sys import glob import subprocess import shutil import netifaces import time import netplan.cli.utils as utils from netplan.configmanager import ConfigManager, ConfigurationError from netplan.cli.sriov import apply_sriov_config from netplan.cli.ovs import OvsDbServerNotRunning, apply_ovs_cleanup OVS_CLEANUP_SERVICE = 'netplan-ovs-cleanup.service' IF_NAMESIZE = 16 class NetplanApply(utils.NetplanCommand): def __init__(self): super().__init__(command_id='apply', description='Apply current netplan config to running system', leaf=True) self.sriov_only = False self.only_ovs_cleanup = False self.state = None # to be filled by the '--state' argument def run(self): # pragma: nocover (covered in autopkgtest) self.parser.add_argument('--sriov-only', action='store_true', help='Only apply SR-IOV related configuration and exit') self.parser.add_argument('--only-ovs-cleanup', action='store_true', help='Only clean up old OpenVSwitch interfaces and exit') self.parser.add_argument('--state', help='Directory containing previous YAML configuration') self.func = self.command_apply self.parse_args() self.run_command() def command_apply(self, run_generate=True, sync=False, exit_on_error=True, state_dir=None): # pragma: nocover config_manager = ConfigManager() if state_dir: self.state = state_dir # For certain use-cases, we might want to only apply specific configuration. # If we only need SR-IOV configuration, do that and exit early. if self.sriov_only: NetplanApply.process_sriov_config(config_manager, exit_on_error) return # If we only need OpenVSwitch cleanup, do that and exit early. elif self.only_ovs_cleanup: NetplanApply.process_ovs_cleanup(config_manager, False, False, exit_on_error) return # if we are inside a snap, then call dbus to run netplan apply instead if "SNAP" in os.environ: # TODO: maybe check if we are inside a classic snap and don't do # this if we are in a classic snap? busctl = shutil.which("busctl") if busctl is None: raise RuntimeError("missing busctl utility") # XXX: DO NOT TOUCH or change this API call, it is used by snapd to communicate # using core20 netplan binary/client/CLI on core18 base systems. Any change # must be agreed upon with the snapd team, so we don't break support for # base systems running older netplan versions. # https://github.com/snapcore/snapd/pull/5915 res = subprocess.call([busctl, "call", "--quiet", "--system", "io.netplan.Netplan", # the service "/io/netplan/Netplan", # the object "io.netplan.Netplan", # the interface "Apply", # the method ]) if res != 0: if exit_on_error: sys.exit(res) elif res == 130: raise PermissionError( "failed to communicate with dbus service") else: raise RuntimeError( "failed to communicate with dbus service: error %s" % res) else: return ovs_cleanup_service = '/run/systemd/system/netplan-ovs-cleanup.service' old_files_networkd = bool(glob.glob('/run/systemd/network/*netplan-*')) old_ovs_glob = glob.glob('/run/systemd/system/netplan-ovs-*') # Ignore netplan-ovs-cleanup.service, as it can always be there if ovs_cleanup_service in old_ovs_glob: old_ovs_glob.remove(ovs_cleanup_service) old_files_ovs = bool(old_ovs_glob) old_nm_glob = glob.glob('/run/NetworkManager/system-connections/netplan-*') nm_ifaces = utils.nm_interfaces(old_nm_glob, netifaces.interfaces()) old_files_nm = bool(old_nm_glob) generator_call = [] generate_out = None if 'NETPLAN_PROFILE' in os.environ: generator_call.extend(['valgrind', '--leak-check=full']) generate_out = subprocess.STDOUT generator_call.append(utils.get_generator_path()) if run_generate and subprocess.call(generator_call, stderr=generate_out) != 0: if exit_on_error: sys.exit(os.EX_CONFIG) else: raise ConfigurationError("the configuration could not be generated") devices = netifaces.interfaces() # Re-start service when # 1. We have configuration files for it # 2. Previously we had config files for it but not anymore # Ideally we should compare the content of the *netplan-* files before and # after generation to minimize the number of re-starts, but the conditions # above works too. restart_networkd = bool(glob.glob('/run/systemd/network/*netplan-*')) if not restart_networkd and old_files_networkd: restart_networkd = True restart_ovs_glob = glob.glob('/run/systemd/system/netplan-ovs-*') # Ignore netplan-ovs-cleanup.service, as it can always be there if ovs_cleanup_service in restart_ovs_glob: restart_ovs_glob.remove(ovs_cleanup_service) restart_ovs = bool(restart_ovs_glob) if not restart_ovs and old_files_ovs: # OVS is managed via systemd units restart_networkd = True restart_nm_glob = glob.glob('/run/NetworkManager/system-connections/netplan-*') nm_ifaces.update(utils.nm_interfaces(restart_nm_glob, devices)) restart_nm = bool(restart_nm_glob) if not restart_nm and old_files_nm: restart_nm = True # Running 'systemctl daemon-reload' will re-run the netplan systemd generator, # so let's make sure we only run it iff we're willing to run 'netplan generate' if run_generate: utils.systemctl_daemon_reload() # stop backends if restart_networkd: logging.debug('netplan generated networkd configuration changed, reloading networkd') # Clean up any old netplan related OVS ports/bonds/bridges, if applicable NetplanApply.process_ovs_cleanup(config_manager, old_files_ovs, restart_ovs, exit_on_error) wpa_services = ['netplan-wpa-*.service'] # Historically (up to v0.98) we had netplan-wpa@*.service files, in case of an # upgraded system, we need to make sure to stop those. if utils.systemctl_is_active('netplan-wpa@*.service'): wpa_services.insert(0, 'netplan-wpa@*.service') utils.systemctl('stop', wpa_services, sync=sync) else: logging.debug('no netplan generated networkd configuration exists') if restart_nm: logging.debug('netplan generated NM configuration changed, restarting NM') if utils.nm_running(): # restarting NM does not cause new config to be applied, need to shut down devices first for device in devices: if device not in nm_ifaces: continue # do not touch this interface # ignore failures here -- some/many devices might not be managed by NM try: utils.nmcli(['device', 'disconnect', device]) except subprocess.CalledProcessError: pass utils.systemctl_network_manager('stop', sync=sync) else: logging.debug('no netplan generated NM configuration exists') # Refresh devices now; restarting a backend might have made something appear. devices = netifaces.interfaces() # evaluate config for extra steps we need to take (like renaming) # for now, only applies to non-virtual (real) devices. config_manager.parse() changes = NetplanApply.process_link_changes(devices, config_manager) # delete virtual interfaces that have been defined in a previous state # but are not configured anymore in the current YAML if self.state: cm = ConfigManager(self.state) cm.parse() # get previous configuration state prev_links = cm.virtual_interfaces.keys() curr_links = config_manager.virtual_interfaces.keys() NetplanApply.clear_virtual_links(prev_links, curr_links, devices) # if the interface is up, we can still apply some .link file changes # but we cannot apply the interface rename via udev, as it won't touch # the interface name, if it was already renamed once (e.g. during boot), # because of the NamePolicy=keep default: # https://www.freedesktop.org/software/systemd/man/systemd.net-naming-scheme.html devices = netifaces.interfaces() for device in devices: logging.debug('netplan triggering .link rules for %s', device) try: subprocess.check_call(['udevadm', 'test-builtin', 'net_setup_link', '/sys/class/net/' + device], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) subprocess.check_call(['udevadm', 'test', '/sys/class/net/' + device], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except subprocess.CalledProcessError: logging.debug('Ignoring device without syspath: %s', device) devices_after_udev = netifaces.interfaces() # apply some more changes manually for iface, settings in changes.items(): # rename non-critical network interfaces new_name = settings.get('name') if new_name: if len(new_name) >= IF_NAMESIZE: logging.warning('Interface name {} is too long. {} will not be renamed'.format(new_name, iface)) continue if iface in devices and new_name in devices_after_udev: logging.debug('Interface rename {} -> {} already happened.'.format(iface, new_name)) continue # re-name already happened via 'udevadm test' # bring down the interface, using its current (matched) interface name subprocess.check_call(['ip', 'link', 'set', 'dev', iface, 'down'], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # rename the interface to the name given via 'set-name' subprocess.check_call(['ip', 'link', 'set', 'dev', iface, 'name', settings.get('name')], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) # Reloading of udev rules happens during 'netplan generate' already # subprocess.check_call(['udevadm', 'control', '--reload-rules']) subprocess.check_call(['udevadm', 'trigger', '--attr-match=subsystem=net']) subprocess.check_call(['udevadm', 'settle']) # apply any SR-IOV related changes, if applicable NetplanApply.process_sriov_config(config_manager, exit_on_error) # (re)set global regulatory domain if os.path.exists('/run/systemd/system/netplan-regdom.service'): utils.systemctl('start', ['netplan-regdom.service']) # (re)start backends if restart_networkd: netplan_wpa = [os.path.basename(f) for f in glob.glob('/run/systemd/system/*.wants/netplan-wpa-*.service')] # exclude the special 'netplan-ovs-cleanup.service' unit netplan_ovs = [os.path.basename(f) for f in glob.glob('/run/systemd/system/*.wants/netplan-ovs-*.service') if not f.endswith('/' + OVS_CLEANUP_SERVICE)] # Run 'systemctl start' command synchronously, to avoid race conditions # with 'oneshot' systemd service units, e.g. netplan-ovs-*.service. try: utils.networkctl_reload() utils.networkctl_reconfigure(utils.networkd_interfaces()) except subprocess.CalledProcessError: # (re-)start systemd-networkd if it is not running, yet logging.warning('Falling back to a hard restart of systemd-networkd.service') utils.systemctl('restart', ['systemd-networkd.service'], sync=True) # 1st: execute OVS cleanup, to avoid races while applying OVS config utils.systemctl('start', [OVS_CLEANUP_SERVICE], sync=True) # 2nd: start all other services utils.systemctl('start', netplan_wpa + netplan_ovs, sync=True) if restart_nm: # Flush all IP addresses of NM managed interfaces, to avoid NM creating # new, non netplan-* connection profiles, using the existing IPs. for iface in utils.nm_interfaces(restart_nm_glob, devices): utils.ip_addr_flush(iface) # clear NM state, especially the [device].managed=true config, as that might have been # re-set via an udev rule setting "NM_UNMANAGED=1" shutil.rmtree('/run/NetworkManager/devices', ignore_errors=True) utils.systemctl_network_manager('start', sync=sync) if sync: # 'nmcli' could be /usr/bin/nmcli or # /snap/bin/nmcli -> /snap/bin/network-manager.nmcli cmd = ['nmcli', 'general', 'status'] # wait a bit for 'connected (site/local-only)' or # 'connected' to appear in 'nmcli general' STATE for _ in range(10): out = subprocess.run(cmd, capture_output=True, text=True) # Handle nmcli's "not running" return code (8) gracefully, # giving some more time for NetworkManager startup if out.returncode == 8: time.sleep(1) continue if '\nconnected' in str(out.stdout): break time.sleep(0.5) @staticmethod def is_composite_member(composites, phy): """ Is this physical interface a member of a 'composite' virtual interface? (bond, bridge) """ for composite in composites: for _, settings in composite.items(): if not type(settings) is dict: continue members = settings.get('interfaces', []) for iface in members: if iface == phy: return True return False @staticmethod def clear_virtual_links(prev_links, curr_links, devices=[]): """ Calculate the delta of virtual links. And remove the links that were dropped from the YAML config, if they were not dropped by the backend already. We can make use of the netplan netdef ids, as those equal the interface name for virtual links. """ if not devices: logging.warning('Cannot clear virtual links: no network interfaces provided.') return [] dropped_interfaces = list(set(prev_links) - set(curr_links)) # some interfaces might have been cleaned up already, e.g. by the # NetworkManager backend interfaces_to_clear = list(set(dropped_interfaces).intersection(devices)) for link in interfaces_to_clear: try: cmd = ['ip', 'link', 'delete', 'dev', link] subprocess.check_call(cmd) except subprocess.CalledProcessError: logging.warning('Could not delete interface {}'.format(link)) return dropped_interfaces @staticmethod def process_link_changes(interfaces, config_manager: ConfigManager): # pragma: nocover (covered in autopkgtest) """ Go through the pending changes and pick what needs special handling. Only applies to non-critical interfaces which can be safely updated. """ changes = {} composite_interfaces = [config_manager.bridges, config_manager.bonds] # Find physical interfaces which need a rename # But do not rename virtual interfaces for netdef in config_manager.physical_interfaces.values(): newname = netdef.set_name if not newname: continue # Skip if no new name needs to be set if not netdef.has_match: continue # Skip if no match for current name is given if NetplanApply.is_composite_member(composite_interfaces, netdef.id): logging.debug('Skipping composite member {}'.format(netdef.id)) # do not rename members of virtual devices. MAC addresses # may be the same for all interface members. continue # Find current name of the interface, according to match conditions and globs (name, mac, driver) current_iface_name = utils.find_matching_iface(interfaces, netdef) if not current_iface_name: logging.warning('Cannot find unique matching interface for {}'.format(netdef.id)) continue if current_iface_name == newname: # Skip interface if it already has the correct name logging.debug('Skipping correctly named interface: {}'.format(newname)) continue if netdef.critical: # Skip interfaces defined as critical, as we should not take them down in order to rename logging.warning('Cannot rename {} ({} -> {}) at runtime (needs reboot), due to being critical' .format(netdef.id, current_iface_name, newname)) continue # record the interface rename change changes[current_iface_name] = {'name': newname} logging.debug('Link changes: {}'.format(changes)) return changes @staticmethod def process_sriov_config(config_manager, exit_on_error=True): # pragma: nocover (covered in autopkgtest) try: apply_sriov_config(config_manager) except utils.config_errors as e: logging.error(str(e)) if exit_on_error: sys.exit(1) @staticmethod def process_ovs_cleanup(config_manager, ovs_old, ovs_current, exit_on_error=True): # pragma: nocover (autopkgtest) try: apply_ovs_cleanup(config_manager, ovs_old, ovs_current) except (OSError, RuntimeError) as e: logging.error(str(e)) if exit_on_error: sys.exit(1) except OvsDbServerNotRunning as e: logging.warning('Cannot call Open vSwitch: {}.'.format(e)) PK ! �2�9c c commands/ip.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018 Canonical, Ltd. # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. '''netplan ip command line''' import logging import os import sys import subprocess from subprocess import CalledProcessError import netplan.cli.utils as utils lease_path = { 'networkd': { 'pattern': 'run/systemd/netif/leases/{lease_id}', 'method': 'ifindex', }, 'NetworkManager': { 'pattern': 'var/lib/NetworkManager/internal-{lease_id}-{interface}.lease', 'method': 'nm_connection', }, } class NetplanIp(utils.NetplanCommand): def __init__(self): super().__init__(command_id='ip', description='Retrieve IP information from the system', leaf=False) def run(self): self.command_leases = NetplanIpLeases() # subcommand: leases p_ip_leases = self.subparsers.add_parser('leases', help='Display IP leases', add_help=False) p_ip_leases.set_defaults(func=self.command_leases.run, commandclass=self.command_leases) self.parse_args() self.run_command() class NetplanIpLeases(utils.NetplanCommand): def __init__(self): super().__init__(command_id='ip leases', description='Display IP leases', leaf=True) def run(self): self.parser.add_argument('interface', help='Interface for which to display IP lease settings.') self.parser.add_argument('--root-dir', help='Search for configuration files in this root directory instead of /') self.func = self.command_ip_leases self.parse_args() self.run_command() def command_ip_leases(self): if self.interface == 'help': # pragma: nocover (covered in autopkgtest) self.print_usage() def find_lease_file(mapping): def lease_method_ifindex(): ifindex_f = os.path.join('/sys/class/net', self.interface, 'ifindex') try: with open(ifindex_f) as f: return f.readlines()[0].strip() except Exception as e: logging.debug('Cannot read file %s: %s', ifindex_f, str(e)) raise def lease_method_nm_connection(): # FIXME: handle older versions of NM where 'nmcli dev show' doesn't exist try: nmcli_dev_out = utils.nmcli_out(['dev', 'show', self.interface]) for line in nmcli_dev_out.splitlines(): if 'GENERAL.CONNECTION' in line: conn_id = line.split(':')[1].rstrip().strip() nmcli_con_out = utils.nmcli_out(['con', 'show', 'id', conn_id]) for line in nmcli_con_out.splitlines(): if 'connection.uuid' in line: return line.split(':')[1].rstrip().strip() except Exception as e: raise Exception('Could not find a NetworkManager connection for the interface: %s' % str(e)) raise Exception('Could not find a NetworkManager connection for the interface') lease_pattern = lease_path[mapping['backend']]['pattern'] lease_method = lease_path[mapping['backend']]['method'] try: lease_id = eval("lease_method_" + lease_method)() # We found something to build the path to the lease file with, # at this point we may have something to look at; but if not, # we'll rely on open() throwing an error. # This might happen if networkd doesn't use DHCP for the interface, # for instance. path = os.path.join('/', os.path.abspath(self.root_dir) if self.root_dir else "", lease_pattern.format(interface=self.interface, lease_id=lease_id)) # Fallback to 'dhclient' if no lease of NetworkManager's # internal DHCP client is found if not os.path.isfile(path): path = path.replace('NetworkManager/internal-', 'NetworkManager/dhclient-') with open(path) as f: for line in f.readlines(): print(line.rstrip()) except Exception as e: print("No lease found for interface '%s': %s" % (self.interface, str(e)), file=sys.stderr) sys.exit(1) argv = [utils.get_generator_path()] if self.root_dir: argv += ['--root-dir', self.root_dir] argv += ['--mapping', self.interface] # Extract out of the generator our mapping in a dict. logging.debug('command ip leases: running %s', argv) try: out = subprocess.check_output(argv, text=True) except CalledProcessError: # pragma: nocover (better be covered in autopkgtest) print("No lease found for interface '%s' (not managed by Netplan)" % self.interface, file=sys.stderr) sys.exit(1) mapping = {} mapping_s = out.split(',') for keyvalue in mapping_s: key, value = keyvalue.strip().split('=') mapping[key] = value find_lease_file(mapping) PK ! �q�� � commands/get.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2020 Canonical, Ltd. # Author: Lukas Märdian <lukas.maerdian@canonical.com> # # 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; version 3. # # 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/>. '''netplan get command line''' import sys import io import tempfile import re import netplan.cli.utils as utils import netplan.libnetplan as libnetplan class NetplanGet(utils.NetplanCommand): def __init__(self): super().__init__(command_id='get', description='Get a setting by specifying a nested key like "ethernets.eth0.addresses", or "all"', leaf=True) def run(self): self.parser.add_argument('key', type=str, nargs='?', default='all', help='The nested key in dotted format') self.parser.add_argument('--root-dir', default='/', help='Read configuration files from this root directory instead of /') self.func = self.command_get self.parse_args() self.run_command() def dump_state(self, key, np_state, output_file): if key == 'all': np_state.dump_yaml(output_file=output_file) return if not key.startswith('network'): key = '.'.join(('network', key)) # Replace the '.' with '\t' but not at '\.' via negative lookbehind expression key = re.sub(r'(?<!\\)\.', '\t', key).replace(r'\.', '.') with tempfile.NamedTemporaryFile() as tmp_in: np_state.dump_yaml(output_file=tmp_in) libnetplan.dump_yaml_subtree(key, tmp_in, output_file=output_file) def command_get(self): parser = libnetplan.Parser() parser.load_yaml_hierarchy(rootdir=self.root_dir) np_state = libnetplan.State() np_state.import_parser_results(parser) try: sys.stdout.fileno() output_file = sys.stdout # pragma: nocover except io.UnsupportedOperation: # Test environment detected, using a buffer file output_file = tempfile.TemporaryFile() self.dump_state(self.key, np_state, output_file) if output_file != sys.stdout: output_file.flush() output_file.seek(0) sys.stdout.write(output_file.read().decode('utf-8')) output_file.close() PK ! ?�[/� � commands/sriov_rebind.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2022 Canonical, Ltd. # Author: Lukas Märdian <slyon@ubuntu.com> # # 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; version 3. # # 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/>. '''netplan SR-IOV rebind command line''' import logging import netplan.cli.utils as utils from netplan.cli.sriov import PCIDevice, bind_vfs, _get_pci_slot_name class NetplanSriovRebind(utils.NetplanCommand): def __init__(self): super().__init__(command_id='rebind', description='Rebind SR-IOV virtual functions of given physical functions to their driver', leaf=True) def run(self): self.parser.add_argument('netdevs', type=str, nargs='*', default=[], help='Space separated list of PF interface names') self.func = self.command_rebind self.parse_args() self.run_command() def command_rebind(self): """Bind virtual functions of SR-IOV devices to their corresponding driver after eswitch mode was changed""" for iface in self.netdevs: pci_addr = _get_pci_slot_name(iface) pcidev = PCIDevice(pci_addr) if not pcidev.is_pf: logging.warning('{} does not seem to be a SR-IOV physical function'.format(iface)) continue bound_vfs = bind_vfs(pcidev.vfs, pcidev.driver) logging.info('{}: bound {} VFs'.format(pcidev, len(bound_vfs))) PK ! P�e�� � core.pynu �[��� #!/usr/bin/python3 # # Copyright (C) 2018 Canonical, Ltd. # Author: Martin Pitt <martin.pitt@ubuntu.com> # Author: Mathieu Trudel-Lapierre <mathieu.trudel-lapierre@canonical.com> # # 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; version 3. # # 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/>. '''netplan command line''' import logging import os import netplan.cli.utils as utils FALLBACK_PATH = '/usr/bin:/snap/bin' class Netplan(utils.NetplanCommand): def __init__(self): super().__init__(command_id='', description='Network configuration in YAML', leaf=False) os.environ.update({ 'LC_ALL': 'C', 'PATH': os.getenv('PATH', FALLBACK_PATH)}) def parse_args(self): import netplan.cli.commands self._import_subcommands(netplan.cli.commands) super().parse_args() def main(self): self.parse_args() if self.debug: logging.basicConfig(level=logging.DEBUG, format='%(levelname)s:%(message)s') os.environ['G_MESSAGES_DEBUG'] = 'all' else: logging.basicConfig(level=logging.INFO, format='%(message)s') self.run_command() PK ! A�&&� � joomla.phpnu �[��� #!/usr/bin/env php <?php /** * @package Joomla.Cli * * @copyright (C) 2017 Open Source Matters, Inc. <https://www.joomla.org> * @license GNU General Public License version 2 or later; see LICENSE.txt */ // We are a valid entry point. const _JEXEC = 1; // Define the application's minimum supported PHP version as a constant so it can be referenced within the application. const JOOMLA_MINIMUM_PHP = '8.1.0'; if (version_compare(PHP_VERSION, JOOMLA_MINIMUM_PHP, '<')) { echo 'Sorry, your PHP version is not supported.' . PHP_EOL; echo 'Your command line php needs to be version ' . JOOMLA_MINIMUM_PHP . ' or newer to run the Joomla! CLI Tools' . PHP_EOL; echo 'The version of PHP currently running this code, at the command line, is PHP version ' . PHP_VERSION . '.' . PHP_EOL; echo 'Please note, the version of PHP running your commands here, may be different to the version that is used by '; echo 'your web server to run the Joomla! Web Application.' . PHP_EOL; exit; } // Load system defines if (file_exists(\dirname(__DIR__) . '/defines.php')) { require_once \dirname(__DIR__) . '/defines.php'; } if (!\defined('_JDEFINES')) { \define('JPATH_BASE', \dirname(__DIR__)); require_once JPATH_BASE . '/includes/defines.php'; } // Check for presence of vendor dependencies not included in the git repository if (!file_exists(JPATH_LIBRARIES . '/vendor/autoload.php') || !is_dir(JPATH_ROOT . '/media/vendor')) { echo 'It looks like you are trying to run Joomla! from our git repository.' . PHP_EOL; echo 'To do so requires you complete a couple of extra steps first.' . PHP_EOL; echo 'Please see https://docs.joomla.org/Special:MyLanguage/J5.x:Setting_Up_Your_Local_Environment for further details.' . PHP_EOL; exit; } // Check if installed if ( !file_exists(JPATH_CONFIGURATION . '/configuration.php') || (filesize(JPATH_CONFIGURATION . '/configuration.php') < 10) ) { echo 'Install Joomla to run cli commands' . PHP_EOL; echo 'To install Joomla from the cli run the following command:' . PHP_EOL; echo 'php installation/joomla.php install' . PHP_EOL; exit; } // Get the framework. require_once JPATH_BASE . '/includes/framework.php'; // Boot the DI container $container = \Joomla\CMS\Factory::getContainer(); /* * Alias the session service keys to the CLI session service as that is the primary session backend for this application * * In addition to aliasing "common" service keys, we also create aliases for the PHP classes to ensure autowiring objects * is supported. This includes aliases for aliased class names, and the keys for aliased class names should be considered * deprecated to be removed when the class name alias is removed as well. */ $container->alias('session', 'session.cli') ->alias('JSession', 'session.cli') ->alias(\Joomla\CMS\Session\Session::class, 'session.cli') ->alias(\Joomla\Session\Session::class, 'session.cli') ->alias(\Joomla\Session\SessionInterface::class, 'session.cli'); $app = \Joomla\CMS\Factory::getContainer()->get(\Joomla\Console\Application::class); \Joomla\CMS\Factory::$application = $app; $app->execute(); PK ! �V� index.htmlnu �[��� <!DOCTYPE html><title></title> PK ! enabled_by_maint/opcachenu �[��� PK ! H enabled_by_maint/zipnu �[��� PK ! � enabled_by_maint/sysvsemnu �[��� PK ! � enabled_by_maint/curlnu �[��� PK ! enabled_by_maint/iconvnu �[��� PK ! _ enabled_by_maint/readlinenu �[��� PK ! � enabled_by_maint/sysvmsgnu �[��� PK ! � enabled_by_maint/pdo_mysqlnu �[��� PK ! : enabled_by_maint/exifnu �[��� PK ! enabled_by_maint/shmopnu �[��� PK ! � enabled_by_maint/gettextnu �[��� PK ! enabled_by_maint/ftpnu �[��� PK ! Q enabled_by_maint/pdonu �[��� PK ! � enabled_by_maint/mysqlndnu �[��� PK ! � enabled_by_maint/fileinfonu �[��� PK ! & enabled_by_maint/pharnu �[��� PK ! k enabled_by_maint/simplexmlnu �[��� PK ! � enabled_by_maint/ctypenu �[��� PK ! � enabled_by_maint/xmlwriternu �[��� PK ! E enabled_by_maint/domnu �[��� PK ! � enabled_by_maint/socketsnu �[��� PK ! � enabled_by_maint/gdnu �[��� PK ! enabled_by_maint/ffinu �[��� PK ! X enabled_by_maint/mbstringnu �[��� PK ! � enabled_by_maint/sysvshmnu �[��� PK ! � enabled_by_maint/mysqlinu �[��� PK ! 0 enabled_by_maint/calendarnu �[��� PK ! y enabled_by_maint/posixnu �[��� PK ! � enabled_by_maint/xmlnu �[��� PK ! enabled_by_maint/xmlreadernu �[��� PK ! M enabled_by_maint/xslnu �[��� PK ! � enabled_by_maint/tokenizernu �[��� PK ! � enabled_by_maint/intlnu �[��� PK ! �8�� � ovs.pynu �[��� PK ! ��](�F �F �) sriov.pynu �[��� PK ! sx9Ň& �& q utils.pynu �[��� PK ! � � � ė __init__.pynu �[��� PK ! �E�� � �� commands/info.pynu �[��� PK ! &�Fx x �� commands/set.pynu �[��� PK ! $��)� � q� commands/generate.pynu �[��� PK ! �P+'� � �� commands/try_command.pynu �[��� PK ! ��~��k �k �� commands/status.pynu �[��� PK ! I��O �O \U commands/migrate.pynu �[��� PK ! Fﱗ� � �� commands/__init__.pynu �[��� PK ! �$<�O �O �� commands/apply.pynu �[��� PK ! �2�9c c �� commands/ip.pynu �[��� PK ! �q�� � ` commands/get.pynu �[��� PK ! ?�[/� � � commands/sriov_rebind.pynu �[��� PK ! P�e�� � o' core.pynu �[��� PK ! A�&&� � 3. joomla.phpnu �[��� PK ! �V� �: index.htmlnu �[��� PK 3 3 � N;
| ver. 1.1 | |
.
| PHP 8.4.18 | Ð“ÐµÐ½ÐµÑ€Ð°Ñ†Ð¸Ñ Ñтраницы: 0.01 |
proxy
|
phpinfo
|
ÐаÑтройка