summaryrefslogtreecommitdiff
path: root/src/helpers/vyos_net_name
blob: 8c09924146a04bfa2dcab4079d681261e7618135 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
#!/usr/bin/env python3
#
# Copyright (C) 2021-2023 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 tempfile
import threading
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.migrator import VirtualMigrator

vyos_udev_dir = directories['vyos_udev_dir']
vyos_log_dir = '/run/udev/log'
vyos_log_file = os.path.join(vyos_log_dir, 'vyos-net-name')

config_path = '/opt/vyatta/etc/config/config.boot'

lock = threading.Lock()

try:
    os.mkdir(vyos_log_dir)
except FileExistsError:
    pass

logging.basicConfig(filename=vyos_log_file, level=logging.DEBUG)

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:
        logging.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:
        logging.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:
        logging.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:
        logging.critical(f"OSError {e}")
        exit(1)

    try:
        config = ConfigTree(config_file)
    except Exception:
        try:
            logging.debug(f"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)
                virtual_migration = VirtualMigrator(fp.name)
                virtual_migration.run()
                with open(fp.name) as fd:
                    config_file = fd.read()

            config = ConfigTree(config_file)

        except Exception as e:
            logging.critical(f"ConfigTree error: {e}")

    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):
                logging.warning(f"no 'hw-id' entry for {intf}")
                continue
            hwid = config.return_value(path)
            if hwid in list(interfaces):
                logging.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):
                logging.warning(f"no 'hw-id' entry for {intf}")
                continue
            hwid = config.return_value(path)
            if hwid in list(interfaces):
                logging.warning(f"multiple entries for {hwid}: {interfaces[hwid]}, {intf}")
                continue
            interfaces[hwid] = intf

    logging.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:
            logging.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
    """
    logging.info(f"lookup {intf_name}, {hwid}")
    interfaces = get_configfile_interfaces()
    logging.debug(f"config file interfaces are {interfaces}")

    if hwid in list(interfaces):
        logging.info(f"use mapping from config file: '{hwid}' -> '{interfaces[hwid]}'")
        return interfaces[hwid]

    add_assigned_interfaces(interfaces)
    logging.debug(f"adding assigned interfaces: {interfaces}")

    if predefined:
        newname = predefined
        logging.info(f"predefined interface name for '{intf_name}' is '{newname}'")
    else:
        newname = get_biosdevname(intf_name)
        logging.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)

    logging.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 len(argv) > 3:
    predef_name = argv[3]
else:
    predef_name = ''

lock.acquire()
if not boot_configuration_complete():
    res = on_boot_event(argv[1], argv[2], predefined=predef_name)
    logging.debug(f"on boot, returned name is {res}")
    print(res)
else:
    logging.debug("boot configuration complete")
lock.release()