Coverage for python/univention/config_registry/interfaces.py: 100%

1#!/usr/bin/python3

3# Univention Common Python Library

5# SPDX-FileCopyrightText: 2010-2026 Univention GmbH

6# SPDX-License-Identifier: AGPL-3.0-only

8"""Handle UCR network configuration."""

10# pylint: disable-msg=W0142,C0103,R0201,R0904

12import re

13from collections.abc import Callable, Iterator, Mapping

14from functools import wraps

15from ipaddress import IPv4Address, IPv4Interface, IPv6Address, IPv6Interface, _IPAddressBase

16from sys import maxsize

17from typing import Any

19from univention.config_registry.backend import ConfigRegistry, ReadOnlyConfigRegistry

22__all__ = ['RE_IFACE', 'Interfaces', 'cmp_alnum', 'forgiving']

24_SKIP = {

25 'interfaces/handler',

26 'interfaces/primary',

27 'interfaces/restart/auto',

28}

29RE_IFACE = re.compile(r'''^

30 (?!(?:%s)$) # _SKIP

31 interfaces/ # prefix

32 ([^/]+)/ # iface name

33 (

34 (?:ipv6/([0-9A-Za-z]+)/)? # IPv6 name

35 .*) # suffix

36 $''' % ('|'.join(_SKIP)), re.VERBOSE)

39def forgiving(translation: dict[type[Exception], Any] | None = None) -> Callable[[Callable], Callable]:

40 """

41 Decorator to translate exceptions into return values.

43 :param translation: Mapping from Exception class to return value.

44 """

45 if translation is None:

46 translation = {}

48 def decorator(func: Callable) -> Callable:

49 """Wrap function and translate exceptions."""

51 @wraps(func)

52 def inner(self, *args, **kwargs):

53 """Run function and translate exceptions."""

54 try:

55 return func(self, *args, **kwargs)

56 except Exception as ex:

57 best = None

58 for cls in translation.keys():

59 if isinstance(ex, cls) and (best is None or issubclass(cls, best)):

60 best = cls

61 if best:

62 return translation[best]

63 raise

65 return inner

67 return decorator

70forgiving_addr = forgiving({ValueError: False, KeyError: None})

71"""Decorator to translate errors from IP address parsing to `None` instead of raising an exception."""

74def cmp_alnum(value: str) -> tuple:

75 """

76 Sort value split by digits / non-digits.

78 :param value: The value to sort.

79 :returns: value split into tuple.

80 """

81 value = str(value)

82 key = []

83 for num, text in RE_ALNUM.findall(value):

84 key.append(int(num or maxsize))

85 key.append(text)

86 return tuple(key)

89RE_ALNUM = re.compile(r'([0-9]+)|([^0-9]+)')

92class _Iface(dict):

93 """Single network interface."""

95 def __init__(self, *args, **kwargs):

96 dict.__init__(self, *args, **kwargs)

97 self.ipv6_names = set()

99 @property

100 def name(self) -> str:

101 """Return interface name."""

102 return self['name'].replace('_', ':')

103

104 @property # type: ignore

105 @forgiving({KeyError: maxsize, ValueError: maxsize})

106 def order(self) -> int:

107 """Return interface order."""

108 return int(self['order'])

109

110 @property

111 def type(self) -> str:

112 """Return interface handler."""

113 return self.get('type', '')

114

115 @property

116 def start(self) -> bool:

117 """Return automatic interface start."""

118 return ConfigRegistry().is_true(value=self.get('start', '1'))

119

120 @property # type: ignore

121 @forgiving_addr

122 def network(self) -> IPv4Address:

123 """Return network address."""

124 return IPv4Address('%(network)s' % self)

125

126 @property # type: ignore

127 @forgiving_addr

128 def broadcast(self) -> IPv4Address:

129 """Return broadcast address."""

130 return IPv4Address('%(broadcast)s' % self)

131

132 @forgiving_addr

133 def ipv4_address(self) -> IPv4Interface:

134 """Return IPv4 address."""

135 return IPv4Interface('%(address)s/%(netmask)s' % self)

136

137 @forgiving_addr

138 def ipv6_address(self, name: str = 'default') -> IPv6Interface:

139 """Return IPv6 address."""

140 key = '%%(ipv6/%s/address)s/%%(ipv6/%s/prefix)s' % (name, name)

141 return IPv6Interface(key % self)

142

143 @property

144 def routes(self) -> Iterator[str]:

145 """Return interface routes."""

146 for k, v in sorted(self.items()):

147 if not k.startswith('route/'):

148 continue

149 if v.startswith(('host ', 'net ')):

150 yield v

151

152 @property

153 def options(self) -> Iterator[str]:

154 """Return interface options."""

155 for k, v in sorted(self.items()):

156 if not k.startswith('options/'):

157 continue

158 yield v

159

160

161class VengefulConfigRegistry(ConfigRegistry):

162 """

163 Instance wrapper for Config Registry throwing exceptions.

164

165 :param base_object: UCR instance.

166

167 <https://forge.univention.org/bugzilla/show_bug.cgi?id=28276>

168 <http://stackoverflow.com/questions/1443129/>

169 """

170

171 def __init__(self, base_object):

172 self.__class__ = type(base_object.__class__.__name__, (self.__class__, base_object.__class__), {})

173 self.__dict__ = base_object.__dict__

174

175 def __getitem__(self, key: str) -> str:

176 """

177 Return registry value.

178

179 Compared with :py:meth:`ConfigRegistry.__getitem__` this raises an exception instead of returning `None`.

180

181 :param key: UCR variable name.

182 :returns: the value.

183 :raises: KeyError when the value is not found.

184 """

185 for _reg, registry in self._walk():

186 try:

187 value = registry[key]

188 return value

189 except KeyError:

190 continue

191 raise KeyError(key)

192

193

194class Interfaces:

195 """

196 Handle network interfaces configured by UCR.

197

198 :param ucr: UCR instance.

199 """

200

201 def __init__(self, ucr: Mapping[str, str] | None = None) -> None:

202 if ucr is None:

203 ucr = ConfigRegistry()

204 ucr.load()

205 if isinstance(ucr, ReadOnlyConfigRegistry):

206 ucr = VengefulConfigRegistry(ucr)

207

208 self.primary = ucr.get('interfaces/primary', 'eth0')

209 try:

210 self.ipv4_gateway: IPv4Address | bool | None = IPv4Address("%(gateway)s" % ucr)

211 except KeyError:

212 self.ipv4_gateway = None

213 except ValueError:

214 self.ipv4_gateway = False

215 try:

216 # <https://tools.ietf.org/html/rfc4007#section-11>

217 # As a common notation to specify the scope zone, an

218 # implementation SHOULD support the following format:

219 # <address>%<zone_id>

220 parts = ucr['ipv6/gateway'].rsplit('%', 1)

221 gateway = parts.pop(0)

222 zone_index = parts[0] if parts else None

223 self.ipv6_gateway: IPv6Address | bool | None = IPv6Address("%s" % (gateway,))

224 self.ipv6_gateway_zone_index = zone_index

225 except KeyError:

226 self.ipv6_gateway = None

227 self.ipv6_gateway_zone_index = None

228 except ValueError:

229 self.ipv6_gateway = False

230 self.ipv6_gateway_zone_index = None

231

232 self._all_interfaces: dict[str, _Iface] = {}

233 for key, value in ucr.items():

234 if not value:

235 continue

236 match = RE_IFACE.match(key)

237 if not match:

238 continue

239 iface, subkey, ipv6_name = match.groups()

240 data = self._all_interfaces.setdefault(iface, _Iface(name=iface))

241 data[subkey] = value

242 if ipv6_name:

243 data.ipv6_names.add(ipv6_name)

244

245 def _cmp_order(self, iface: _Iface) -> tuple[tuple, tuple]:

246 """

247 Compare interfaces by order.

248

249 :param iface: Other interface.

250 :returns: A tuple to be used as a key for sorting.

251 """

252 return (

253 cmp_alnum(iface.order),

254 cmp_alnum(iface.name),

255 )

256

257 def _cmp_primary(self, iface: _Iface) -> tuple[int, tuple, tuple]:

258 """

259 Compare interfaces by primary.

260

261 :param iface: Other interface.

262 :returns: 3-tuple to be used as a key for sorting.

263 """

264 try:

265 primary = self.primary.index(iface.name)

266 except ValueError:

267 primary = maxsize

268 return (

269 primary,

270 cmp_alnum(iface.order),

271 cmp_alnum(iface.name),

272 )

273

274 def _cmp_name(self, iname: str) -> str:

275 """

276 Compare IPv6 sub-interfaces by name.

277

278 :param name: Interface name.

279 :returns: string used as a key for sorting.

280 """

281 return '' if iname == 'default' else iname

282

283 @property

284 def all_interfaces(self) -> Iterator[tuple[str, _Iface]]:

285 """Yield IPv4 interfaces."""

286 yield from sorted(self._all_interfaces.items(), key=lambda name_iface: self._cmp_order(name_iface[1]))

287

288 @property

289 def ipv4_interfaces(self) -> Iterator[tuple[str, _Iface]]:

290 """Yield IPv4 interfaces."""

291 for name, iface in sorted(self._all_interfaces.items(), key=lambda _name_iface: self._cmp_order(_name_iface[1])):

292 if iface.ipv4_address() is not None:

293 yield (name, iface)

294

295 @property

296 def ipv6_interfaces(self) -> Iterator[tuple[_Iface, str]]:

297 """Yield names of IPv6 interfaces."""

298 for iface in sorted(self._all_interfaces.values(), key=self._cmp_order):

299 for name in sorted(iface.ipv6_names, key=self._cmp_name):

300 if iface.ipv6_address(name):

301 yield (iface, name)

302

303 def get_default_ip_address(self) -> _IPAddressBase | None:

304 """returns the default IP address."""

305 for iface in sorted(self._all_interfaces.values(), key=self._cmp_primary):

306 addr = iface.ipv4_address()

307 if addr:

308 return addr

309 for name in sorted(iface.ipv6_names, key=self._cmp_name):

310 addr = iface.ipv6_address(name)

311 if addr:

312 return addr

313

314 return None

315

316 def get_default_ipv4_address(self) -> IPv4Interface | None:

317 """returns the default IPv4 address."""

318 for iface in sorted(self._all_interfaces.values(), key=self._cmp_primary):

319 addr = iface.ipv4_address()

320 if addr:

321 return addr

322

323 return None

324

325 def get_default_ipv6_address(self) -> IPv6Interface | None:

326 """returns the default IPv6 address."""

327 for iface in sorted(self._all_interfaces.values(), key=self._cmp_primary):

328 for name in sorted(iface.ipv6_names, key=self._cmp_name):

329 addr = iface.ipv6_address(name)

330 if addr:

331 return addr

332

333 return None