T6732: added same as vyos 1x

1 files changed, 276 insertions, 0 deletions

diff --git a/src/helpers/vyos_net_name b/src/helpers/vyos_net_name
new file mode 100644
index 0000000..f5de182
--- /dev/null
+++ b/src/helpers/vyos_net_name
@@ -0,0 +1,276 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2021-2024 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import re
+import time
+import logging
+import logging.handlers
+import tempfile
+from pathlib import Path
+from sys import argv
+
+from vyos.configtree import ConfigTree
+from vyos.defaults import directories
+from vyos.utils.process import cmd
+from vyos.utils.boot import boot_configuration_complete
+from vyos.utils.locking import Lock
+from vyos.migrate import ConfigMigrate
+
+# Define variables
+vyos_udev_dir = directories['vyos_udev_dir']
+config_path = '/opt/vyatta/etc/config/config.boot'
+
+
+def is_available(intfs: dict, intf_name: str) -> bool:
+    """Check if interface name is already assigned"""
+    if intf_name in list(intfs.values()):
+        return False
+    return True
+
+
+def find_available(intfs: dict, prefix: str) -> str:
+    """Find lowest indexed iterface name that is not assigned"""
+    index_list = [
+        int(x.replace(prefix, '')) for x in list(intfs.values()) if prefix in x
+    ]
+    index_list.sort()
+    # find 'holes' in list, if any
+    missing = sorted(set(range(index_list[0], index_list[-1])) - set(index_list))
+    if missing:
+        return f'{prefix}{missing[0]}'
+
+    return f'{prefix}{len(index_list)}'
+
+
+def mod_ifname(ifname: str) -> str:
+    """Check interface with names eX and return ifname on the next format eth{ifindex} - 2"""
+    if re.match('^e[0-9]+$', ifname):
+        intf = ifname.split('e')
+        if intf[1]:
+            if int(intf[1]) >= 2:
+                return 'eth' + str(int(intf[1]) - 2)
+            else:
+                return 'eth' + str(intf[1])
+
+    return ifname
+
+
+def get_biosdevname(ifname: str) -> str:
+    """Use legacy vyatta-biosdevname to query for name
+
+    This is carried over for compatability only, and will likely be dropped
+    going forward.
+    XXX: This throws an error, and likely has for a long time, unnoticed
+    since vyatta_net_name redirected stderr to /dev/null.
+    """
+    intf = mod_ifname(ifname)
+
+    if 'eth' not in intf:
+        return intf
+    if os.path.isdir('/proc/xen'):
+        return intf
+
+    time.sleep(1)
+
+    try:
+        biosname = cmd(f'/sbin/biosdevname --policy all_ethN -i {ifname}')
+    except Exception as e:
+        logger.error(f'biosdevname error: {e}')
+        biosname = ''
+
+    return intf if biosname == '' else biosname
+
+
+def leave_rescan_hint(intf_name: str, hwid: str):
+    """Write interface information reported by udev
+
+    This script is called while the root mount is still read-only. Leave
+    information in /run/udev: file name, the interface; contents, the
+    hardware id.
+    """
+    try:
+        os.mkdir(vyos_udev_dir)
+    except FileExistsError:
+        pass
+    except Exception as e:
+        logger.critical(f'Error creating rescan hint directory: {e}')
+        exit(1)
+
+    try:
+        with open(os.path.join(vyos_udev_dir, intf_name), 'w') as f:
+            f.write(hwid)
+    except OSError as e:
+        logger.critical(f'OSError {e}')
+
+
+def get_configfile_interfaces() -> dict:
+    """Read existing interfaces from config file"""
+    interfaces: dict = {}
+
+    if not os.path.isfile(config_path):
+        # If the case, then we are running off of livecd; return empty
+        return interfaces
+
+    try:
+        with open(config_path) as f:
+            config_file = f.read()
+    except OSError as e:
+        logger.critical(f'OSError {e}')
+        exit(1)
+
+    try:
+        config = ConfigTree(config_file)
+    except Exception:
+        try:
+            logger.debug('updating component version string syntax')
+            # this will update the component version string syntax,
+            # required for updates 1.2 --> 1.3/1.4
+            with tempfile.NamedTemporaryFile() as fp:
+                with open(fp.name, 'w') as fd:
+                    fd.write(config_file)
+                config_migrate = ConfigMigrate(fp.name)
+                if config_migrate.syntax_update_needed():
+                    config_migrate.update_syntax()
+                    config_migrate.write_config()
+                with open(fp.name) as fd:
+                    config_file = fd.read()
+
+            config = ConfigTree(config_file)
+
+        except Exception as e:
+            logger.critical(f'ConfigTree error: {e}')
+            exit(1)
+
+    base = ['interfaces', 'ethernet']
+    if config.exists(base):
+        eth_intfs = config.list_nodes(base)
+        for intf in eth_intfs:
+            path = base + [intf, 'hw-id']
+            if not config.exists(path):
+                logger.warning(f"no 'hw-id' entry for {intf}")
+                continue
+            hwid = config.return_value(path)
+            if hwid in list(interfaces):
+                logger.warning(
+                    f'multiple entries for {hwid}: {interfaces[hwid]}, {intf}'
+                )
+                continue
+            interfaces[hwid] = intf
+
+    base = ['interfaces', 'wireless']
+    if config.exists(base):
+        wlan_intfs = config.list_nodes(base)
+        for intf in wlan_intfs:
+            path = base + [intf, 'hw-id']
+            if not config.exists(path):
+                logger.warning(f"no 'hw-id' entry for {intf}")
+                continue
+            hwid = config.return_value(path)
+            if hwid in list(interfaces):
+                logger.warning(
+                    f'multiple entries for {hwid}: {interfaces[hwid]}, {intf}'
+                )
+                continue
+            interfaces[hwid] = intf
+
+    logger.debug(f'config file entries: {interfaces}')
+
+    return interfaces
+
+
+def add_assigned_interfaces(intfs: dict):
+    """Add interfaces found by previous invocation of udev rule"""
+    if not os.path.isdir(vyos_udev_dir):
+        return
+
+    for intf in os.listdir(vyos_udev_dir):
+        path = os.path.join(vyos_udev_dir, intf)
+        try:
+            with open(path) as f:
+                hwid = f.read().rstrip()
+        except OSError as e:
+            logger.error(f'OSError {e}')
+            continue
+        intfs[hwid] = intf
+
+
+def on_boot_event(intf_name: str, hwid: str, predefined: str = '') -> str:
+    """Called on boot by vyos-router: 'coldplug' in vyatta_net_name"""
+    logger.info(f'lookup {intf_name}, {hwid}')
+    interfaces = get_configfile_interfaces()
+    logger.debug(f'config file interfaces are {interfaces}')
+
+    if hwid in list(interfaces):
+        logger.info(f"use mapping from config file: '{hwid}' -> '{interfaces[hwid]}'")
+        return interfaces[hwid]
+
+    add_assigned_interfaces(interfaces)
+    logger.debug(f'adding assigned interfaces: {interfaces}')
+
+    if predefined:
+        newname = predefined
+        logger.info(f"predefined interface name for '{intf_name}' is '{newname}'")
+    else:
+        newname = get_biosdevname(intf_name)
+        logger.info(f"biosdevname returned '{newname}' for '{intf_name}'")
+
+    if not is_available(interfaces, newname):
+        prefix = re.sub(r'\d+$', '', newname)
+        newname = find_available(interfaces, prefix)
+
+    logger.info(f"new name for '{intf_name}' is '{newname}'")
+
+    leave_rescan_hint(newname, hwid)
+
+    return newname
+
+
+def hotplug_event():
+    # Not yet implemented, since interface-rescan will only be run on boot.
+    pass
+
+
+if __name__ == '__main__':
+    # Set up logging to syslog
+    syslog_handler = logging.handlers.SysLogHandler(address='/dev/log')
+    formatter = logging.Formatter(f'{Path(__file__).name}: %(message)s')
+    syslog_handler.setFormatter(formatter)
+
+    logger = logging.getLogger()
+    logger.addHandler(syslog_handler)
+    logger.setLevel(logging.DEBUG)
+
+    logger.debug(f'Started with arguments: {argv}')
+
+    if len(argv) > 3:
+        predef_name = argv[3]
+    else:
+        predef_name = ''
+
+    lock = Lock('vyos_net_name')
+    # Wait 60 seconds for other running scripts to finish
+    lock.acquire(60)
+
+    if not boot_configuration_complete():
+        res = on_boot_event(argv[1], argv[2], predefined=predef_name)
+        logger.debug(f'on boot, returned name is {res}')
+        print(res)
+    else:
+        logger.debug('boot configuration complete')
+
+    lock.release()
+    logger.debug('Finished')