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
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
|
"""
NETCONF for APK Distributions server:
Network Management System abstraction module for ifupdown-ng
Copyright © 2020 Adélie Software in the Public Benefit, Inc.
Released under the terms of the NCSA license. See the LICENSE file included
with this source distribution for more information.
SPDX-License-Identifier: NCSA
"""
import ipaddress
import logging
import socket
import subprocess
import yaml
from ncserver.base.util import _
LOGGER = logging.getLogger(__name__)
"""The object used for logging informational messages."""
M_ABI_VERSION = 1
"""The ABI version of this NETCONF module."""
M_PREFIX = "nmsa"
"""The XML tag prefix for this module's tags."""
M_NS = "http://netconf.adelielinux.org/ns/netmgmt"
"""The XML namespace for this module."""
M_NAME = "adelie-nms-abstract"
"""The YANG model name for this module."""
M_REVISION = "2020-11-18"
"""The YANG revision date for this module."""
_CONFIG = dict()
"""The internal configuration handle."""
_TRANSACTION = False
"""Determines if a transaction is in progress."""
#############################
# I N T E R N A L #
#############################
def _load_config():
"""Load the current active configuration from /e/n/i."""
global _CONFIG # pylint: disable=W0603
# Won't load during a transaction.
if _TRANSACTION:
return
result = None
try:
result = subprocess.run(['/sbin/ifparse', '-AF',
'yaml-raw'],
stdout=subprocess.PIPE, check=False)
except OSError:
LOGGER.error(_("ifupdown-ng may not be installed properly"))
return
if result.returncode != 0:
LOGGER.error(_("ifparse returned error %d"), result.returncode)
return
rawyaml = result.stdout.decode('utf-8')
_CONFIG = yaml.safe_load(rawyaml)
def _save_unlocked():
"""Save changes to the configuration."""
eni = ""
for iface in _CONFIG.keys():
buf = "iface " + iface + "\n"
for item in _CONFIG[iface]:
for key, val in item.items():
if key == 'auto' and \
(val is True or val.lower()[0] == 't'):
buf = "auto " + iface + "\n" + buf
continue
if isinstance(val, bool):
val = str(val).lower()
buf += " "
buf += str(key) + " " + str(val)
buf += "\n"
eni += buf + "\n"
with open('/etc/network/interfaces', 'w') as conf_file:
# snip last double-\n off
conf_file.write(eni[:-1])
def _save():
"""Save configuration changes, if a transaction is not in progress."""
if _TRANSACTION:
return
_save_unlocked()
def _find_one(iface: str, key: str):
"""Find a single instance of configuration +key+ for +iface+."""
if iface not in _CONFIG.keys():
return None
for item in _CONFIG[iface]:
if key in item.keys():
return item[key]
return None
def _find_many(iface: str, key: str) -> list:
"""Find n instances of configuration +key+ for +iface+."""
ret = list()
if iface not in _CONFIG.keys():
return None
for item in _CONFIG[iface]:
if key in item.keys():
ret.append(item[key])
return ret
def _replace_one(iface: str, key: str, value: str):
"""Replace a single instance of +key+ for +iface+ with +value+."""
iface_cfg = _CONFIG[iface]
for item in iface_cfg:
if key in item.keys():
item[key] = value
_save()
return
iface_cfg.append({key: value})
_save()
return
def _remove_one(iface: str, key: str):
"""Remove a single instance of +key+ from +iface+."""
iface_cfg = _CONFIG[iface]
for item in iface_cfg:
if key in item.keys():
iface_cfg.remove(item)
_save()
return
#############################
# P A R A M E T E R S #
#############################
def get_desc(iface: str, _):
"""Retrieve the description for the specified interface."""
return _find_one(iface, 'netconf-description')
def set_desc(iface: str, _, value: str):
"""Set the description for the specified interface."""
_replace_one(iface, 'netconf-description', value)
def unset_desc(iface: str, _):
"""Unset the description for the specified interface."""
_remove_one(iface, 'netconf-description')
_PARAMETERS = {
# "name": (getter, setter, unsetter)
"description": (get_desc, set_desc, unset_desc)
}
"""Describes all supported parameters and their methods."""
#############################
# P U B L I C A P I #
#############################
def interface_list():
"""Return a list of configured interfaces."""
_load_config()
return tuple(_CONFIG.keys())
def remove_interface(iface: str):
"""Completely remove configuration for +iface+."""
if iface in _CONFIG.keys():
del _CONFIG[iface]
if not _TRANSACTION:
_save()
def begin_transaction():
"""Begin a transaction."""
global _TRANSACTION # pylint: disable=W0603
if _TRANSACTION:
LOGGER.error(_("attempt to nest transactions"))
return
_TRANSACTION = True
def commit():
"""Commit any outstanding operations."""
global _TRANSACTION # pylint: disable=W0603
if not _TRANSACTION:
LOGGER.warning(_("commit when no transaction is in progress"))
_save_unlocked()
_TRANSACTION = False
def get_param(iface: str, parameter: str):
"""Retrieve the parameter for the specified interface."""
_load_config()
if iface not in _CONFIG.keys():
LOGGER.warning(
_("requested parameter %s for non-existent interface %s"),
parameter, iface
)
return None
if parameter not in _PARAMETERS.keys():
LOGGER.error(_("requested non-existent parameter %s for interface %s"),
parameter, iface)
return None
return _PARAMETERS[parameter][0](iface, parameter)
def set_param(iface: str, parameter: str, value):
"""Set the parameter for the specified interface."""
_load_config()
if parameter not in _PARAMETERS.keys():
LOGGER.error(
_("attempted to set non-existent parameter %s for interface %s"),
parameter, iface
)
return
# Allow creation of new interfaces from NETCONF.
if iface not in _CONFIG.keys():
_CONFIG[iface] = list()
_PARAMETERS[parameter][1](iface, paremeter, value)
def unset_param(iface: str, parameter: str):
"""Unset the parameter for the specified interface."""
_load_config()
if iface not in _CONFIG.keys():
LOGGER.warning(
_("attempted to unset parameter %s for non-existent interface %s"),
parameter, iface
)
return
if parameter not in _PARAMETERS.keys():
LOGGER.error(
_("attempted to unset non-existent parameter %s for interface %s"),
parameter, iface
)
return
_PARAMETERS[parameter][2](iface, parameter)
def list_addresses(iface: str) -> list:
"""Retrieve all configured addresses for the specified interface."""
_load_config()
if iface not in _CONFIG.keys():
LOGGER.warning(_("requested addresses for non-existent interface %s"),
iface)
return list()
# Per comment in lif_address_format_cidr, this is the right thing to do.
fallback_prefix = "24"
netmask = _find_one(iface, 'netmask')
if netmask:
net = ipaddress.IPv4Network('0.0.0.0/'+netmask)
fallback_prefix = str(net.prefixlen)
addrs = _find_many(iface, 'address')
def fixup(addr):
if '/' not in addr:
addr = addr + "/" + fallback_prefix
return addr
return list(map(fixup, addrs))
def add_address(iface: str, _type, addr: str, prefix):
"""Add an address of the specified ``type`` to the specified interface."""
_load_config()
if iface not in _CONFIG.keys():
LOGGER.warning(
_("attempted to add address to non-existent interface %s"), iface
)
return
if _type not in (socket.AF_INET, socket.AF_INET6):
LOGGER.error(_("unknown address type %r"), _type)
return
# implement this.
raise NotImplementedError
def remove_address(iface: str, addr: str):
"""Remove an address from the specified interface."""
_load_config()
if iface not in _CONFIG.keys():
LOGGER.warning(
_("attempted to remove address from non-existent interface %s"),
iface
)
return
# implement this.
raise NotImplementedError
# Load immediately when we're loaded so we can go straight to a transaction
# if desired.
_load_config()
|