# SPDX-FileCopyrightText: 2004-2025 Univention GmbH
# SPDX-License-Identifier: AGPL-3.0-only
"""Univention Setup: network configuration abstract base classes"""
import logging
import subprocess
from collections.abc import Sequence
from ipaddress import IPv4Interface, IPv6Interface
from univention.config_registry import ConfigRegistry
from univention.config_registry.interfaces import Interfaces
[docs]
class ChangeSet:
def __init__(self, ucr: ConfigRegistry, profile: dict[str, str], options):
self.ucr = ucr
self.profile = profile
self.options = options
self.ucr_changes: dict[str, str | None] = {}
self.old_interfaces = Interfaces(ucr)
self.logger = logging.getLogger("uss.network.change")
self.update_config(self.only_network_config(profile))
[docs]
@staticmethod
def only_network_config(profile: dict[str, str]) -> dict[str, str | None]:
config: dict[str, str | None] = {}
for key, value in profile.items():
if key.startswith("interfaces/"):
config[key] = value or None
return config
[docs]
def update_config(self, changes: dict[str, str | None]) -> None:
self.ucr_changes.update(changes)
new_ucr = dict(self.ucr.items()) # Bug #33101
new_ucr.update(changes)
self.new_interfaces = Interfaces(new_ucr)
@property
def no_act(self) -> bool:
return self.options.no_act
@property
def old_names(self) -> set[str]:
return {name for name, _iface in self.old_interfaces.all_interfaces}
@property
def new_names(self) -> set[str]:
return {name for name, _iface in self.new_interfaces.all_interfaces}
@property
def old_ipv4s(self) -> list[IPv4Interface]:
return [iface.ipv4_address() for _name, iface in self.old_interfaces.ipv4_interfaces]
@property
def new_ipv4s(self) -> list[IPv4Interface]:
return [iface.ipv4_address() for _name, iface in self.new_interfaces.ipv4_interfaces]
@property
def old_ipv6s(self) -> list[IPv6Interface]:
return [iface.ipv6_address(name) for iface, name in self.old_interfaces.ipv6_interfaces]
@property
def new_ipv6s(self) -> list[IPv6Interface]:
return [iface.ipv6_address(name) for iface, name in self.new_interfaces.ipv6_interfaces]
[docs]
class SkipPhase(Exception):
pass
[docs]
class Phase: # noqa: PLW1641
"""Base-class for all phases."""
priority = 0
def __init__(self, changeset: ChangeSet) -> None:
self.changeset = changeset
self.logger = logging.getLogger("uss.network.phase.%s" % (self,))
def __lt__(self, other: object) -> object:
"""
Order phases by priority.
>>> Phase(None) < Phase(None)
False
>>> Phase(None) <= Phase(None)
True
>>> Phase(None) == Phase(None)
True
>>> Phase(None) != Phase(None)
False
>>> Phase(None) >= Phase(None)
True
>>> Phase(None) > Phase(None)
False
"""
return (self.priority, str(self)) < (other.priority, str(other)) if isinstance(other, Phase) else NotImplemented
def __le__(self, other: object) -> object:
return (self.priority, str(self)) <= (other.priority, str(other)) if isinstance(other, Phase) else NotImplemented
def __eq__(self, other: object) -> bool:
return isinstance(other, Phase) and (self.priority, str(self)) == (other.priority, str(other))
def __ne__(self, other: object) -> bool:
return not self.__eq__(other)
def __ge__(self, other: object) -> object:
return (self.priority, str(self)) >= (other.priority, str(other)) if isinstance(other, Phase) else NotImplemented
def __gt__(self, other: object) -> object:
return (self.priority, str(self)) > (other.priority, str(other)) if isinstance(other, Phase) else NotImplemented
def __str__(self) -> str:
name = self.__class__.__name__
name = name.removeprefix("Phase")
return name
@classmethod
def _check_valid(cls, other: type["Phase"]) -> None:
try:
if not issubclass(other, cls):
raise SkipPhase('Invalid super-class')
if not other.priority:
raise SkipPhase('Missing priority')
# if type(other) is ABCMeta:
# raise SkipPhase('Abstract class')
except TypeError:
raise SkipPhase('Not a class')
[docs]
def check(self) -> None:
"""
Check if the phase should be activated.
Throw SkipPhase to skip this phase.
"""
[docs]
def pre(self) -> None:
"""Called before the changes are applied to UCR."""
[docs]
def post(self) -> None:
"""Called after the changes have been applied to UCR."""
[docs]
def call(self, command: Sequence[str]) -> int:
"""Call external command using subprocess.call(shell=False)."""
self.logger.debug("Running %r", command)
if self.changeset.no_act:
ret = 0
else:
ret = subprocess.call(command)
self.logger.debug("%r returned %d", command, ret)
return ret
