diff --git a/mfd_network_adapter/network_interface/feature/ip/linux.py b/mfd_network_adapter/network_interface/feature/ip/linux.py index 3b62191..116e975 100644 --- a/mfd_network_adapter/network_interface/feature/ip/linux.py +++ b/mfd_network_adapter/network_interface/feature/ip/linux.py @@ -183,30 +183,201 @@ def set_ipv6_autoconf(self, state: State = State.ENABLED) -> None: for cmd in cmds: self._connection.execute_command(cmd) - def release_ip(self, ip_version: IPVersion) -> None: + def _release_with_dhclient(self) -> None: """ - Remove DHCP address. + Release DHCP lease using dhclient. - :param ip_version: IP version to use + :raises IPFeatureException: When dhclient returns an unexpected error """ - # release current lease cmd = f"dhclient -r {self._interface().name}" output = self._connection.execute_command(cmd, expected_return_codes={}) - if output.return_code and "dhcpcd not running" not in output.stdout: + output_text = f"{output.stdout}\n{output.stderr}" + lowered_output = output_text.casefold() + dhclient_not_running = "dhcpcd not running" in lowered_output or "no such process" in lowered_output + if output.return_code and not dhclient_not_running: raise IPFeatureException( f"Unknown error msg returned, while releasing IP on {self._interface().name} - {output.stderr}" ) - # RHEL7 still keeps IP after releasing it, so we need to remove it + def _nm_manages_dhcp(self, ip_version: IPVersion) -> bool: + """ + Check whether NetworkManager manages DHCP for this interface and IP version. + + :param ip_version: IP version for which DHCP management should be validated + :return: True when NetworkManager manages this interface and requested DHCP mode, otherwise False + """ + nm_managed_output = self._connection.execute_command( + f"nmcli -g GENERAL.NM-MANAGED device show {self._interface().name}", expected_return_codes={0, 1} + ) + nm_managed = nm_managed_output.return_code == 0 and nm_managed_output.stdout.strip().casefold() == "yes" + if not nm_managed: + logger.warning(msg=f"NetworkManager does not manage {self._interface().name}; skipping NM release.") + return False + + conn_name_output = self._connection.execute_command( + f"nmcli -g GENERAL.CONNECTION device show {self._interface().name}", + expected_return_codes={0, 1}, + ) + connection_name = conn_name_output.stdout.strip() + connection_active = conn_name_output.return_code == 0 and bool(connection_name) and connection_name != "--" + if not connection_active: + logger.warning( + msg=f"No active NetworkManager connection for {self._interface().name}; skipping NM release." + ) + return False + + methods_output = self._connection.execute_command( + f'nmcli -g ipv4.method,ipv6.method connection show "{connection_name}"', + expected_return_codes={0, 1}, + ) + method_lines = [line.strip().casefold() for line in methods_output.stdout.splitlines()] + v4_method = method_lines[0] if len(method_lines) > 0 else "" + v6_method = method_lines[1] if len(method_lines) > 1 else "" + nm_has_dhcp = methods_output.return_code == 0 and ( + (ip_version is IPVersion.V4 and v4_method == "auto") or (ip_version is IPVersion.V6 and v6_method == "auto") + ) + if not nm_has_dhcp: + logger.warning( + msg=f"NetworkManager is not configured for DHCP on {self._interface().name}; skipping NM release." + ) + return False + + return True + + def _release_with_network_manager(self) -> None: + """ + Release DHCP lease using NetworkManager/nmcli. + + :raises IPFeatureException: When nmcli returns an unexpected error + """ + nmcli_release = self._connection.execute_command( + f"nmcli device disconnect {self._interface().name}", + expected_return_codes={}, + ) + nmcli_output_text = f"{nmcli_release.stdout}\n{nmcli_release.stderr}" + lowered_nmcli_output = nmcli_output_text.casefold() + nmcli_non_fatal = ( + "not active" in lowered_nmcli_output + or "not connected" in lowered_nmcli_output + or "not managed" in lowered_nmcli_output + or "unknown device" in lowered_nmcli_output + or "device disconnect failed" in lowered_nmcli_output + ) + if nmcli_release.return_code and not nmcli_non_fatal: + raise IPFeatureException( + "Unknown error msg returned, while releasing IP with NetworkManager on " + f"{self._interface().name} - {nmcli_release.stderr}" + ) + + def _release_with_wicked(self) -> None: + """ + Release DHCP lease using wicked. + + The method first tries ``ifreload`` and falls back to + ``ifdown --release --no-delete`` when reload fails with a fatal error. + + :raises IPFeatureException: When both wicked commands return unexpected errors + """ + wicked_reload = self._connection.execute_command( + f"wicked ifreload {self._interface().name}", + expected_return_codes={}, + ) + reload_output_text = f"{wicked_reload.stdout}\n{wicked_reload.stderr}" + lowered_reload_output = reload_output_text.casefold() + wicked_non_fatal = ( + "already down" in lowered_reload_output + or "is down" in lowered_reload_output + or "no such device" in lowered_reload_output + or "not configured" in lowered_reload_output + or "no matching interfaces" in lowered_reload_output + ) + if wicked_reload.return_code and not wicked_non_fatal: + wicked_fallback = self._connection.execute_command( + f"wicked ifdown --release --no-delete {self._interface().name}", + expected_return_codes={}, + ) + fallback_output_text = f"{wicked_fallback.stdout}\n{wicked_fallback.stderr}" + lowered_fallback_output = fallback_output_text.casefold() + fallback_non_fatal = ( + "already down" in lowered_fallback_output + or "is down" in lowered_fallback_output + or "no such device" in lowered_fallback_output + or "not configured" in lowered_fallback_output + or "no matching interfaces" in lowered_fallback_output + ) + if wicked_fallback.return_code and not fallback_non_fatal: + raise IPFeatureException( + f"Unknown error msg returned, while releasing IP with wicked on {self._interface().name} - " + f"reload error: {wicked_reload.stderr}; fallback error: {wicked_fallback.stderr}" + ) + + def _detect_release_manager(self, ip_version: IPVersion) -> str: + """ + Detect which DHCP manager can be used to release lease. + + Selection order is: ``dhclient``, ``network_manager``, ``wicked``, ``none``. + + :param ip_version: IP version used to verify NetworkManager DHCP settings + :return: Manager identifier used by ``release_ip`` dispatcher + """ + dhclient_output = self._connection.execute_command("command -v dhclient", expected_return_codes={0, 1}) + if dhclient_output.return_code == 0: + return "dhclient" + + logger.warning(msg=f"dhclient is not available on {self._interface().name}; trying NetworkManager.") + nmcli_output = self._connection.execute_command("command -v nmcli", expected_return_codes={0, 1}) + if nmcli_output.return_code == 0: + if self._nm_manages_dhcp(ip_version): + return "network_manager" + else: + logger.warning(msg=f"nmcli is not available on {self._interface().name}; skipping NM lease release flow.") + + logger.warning(msg=f"NetworkManager DHCP flow unavailable on {self._interface().name}; trying wicked.") + wicked_output = self._connection.execute_command("command -v wicked", expected_return_codes={0, 1}) + if wicked_output.return_code == 0: + return "wicked" + + logger.warning(msg=f"wicked is not available on {self._interface().name}; skipping lease release command.") + return "none" + + def _flush_released_ip(self, ip_version: IPVersion) -> None: + """ + Flush interface IPs as final release fallback. + + :param ip_version: IP version to flush from the interface + """ cmd = f"ip -{ip_version.value} addr flush dev {self._interface().name}" self._connection.execute_command(cmd) + def release_ip(self, ip_version: IPVersion) -> None: + """ + Remove DHCP address. + + :param ip_version: IP version to use + """ + manager = self._detect_release_manager(ip_version) + match manager: + case "dhclient": + self._release_with_dhclient() + case "network_manager": + self._release_with_network_manager() + case "wicked": + self._release_with_wicked() + case _: + pass + + # RHEL7 may keep IP after lease release, so always flush the interface IPs. + self._flush_released_ip(ip_version) + def renew_ip(self) -> None: """Refresh Ip address.""" # release current lease cmd = f"dhclient -r {self._interface().name}" output = self._connection.execute_command(cmd, expected_return_codes={}) - if output.return_code and "dhcpcd not running" not in output.stdout.casefold(): + output_text = f"{output.stdout}\n{output.stderr}" + lowered_output = output_text.casefold() + dhclient_not_running = "dhcpcd not running" in lowered_output or "no such process" in lowered_output + if output.return_code and not dhclient_not_running: raise IPFeatureException( f"Unknown error msg returned, while releasing current lease on {self._interface().name} - " f"{output.stderr}" diff --git a/tests/unit/test_mfd_network_adapter/test_network_interface/test_feature/test_ip/test_ip_linux.py b/tests/unit/test_mfd_network_adapter/test_network_interface/test_feature/test_ip/test_ip_linux.py index 2f2a465..8fbf2a2 100644 --- a/tests/unit/test_mfd_network_adapter/test_network_interface/test_feature/test_ip/test_ip_linux.py +++ b/tests/unit/test_mfd_network_adapter/test_network_interface/test_feature/test_ip/test_ip_linux.py @@ -172,12 +172,15 @@ def test_set_ipv6_autoconf_disabled(self, interface): ) def test_release_ip(self, interface): - interface._connection.execute_command.return_value = ConnectionCompletedProcess( - return_code=0, args="", stdout="", stderr="" - ) + interface._connection.execute_command.side_effect = [ + ConnectionCompletedProcess(return_code=0, args="", stdout="/usr/sbin/dhclient", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ] interface.ip.release_ip(IPVersion.V4) interface._connection.execute_command.assert_has_calls( [ + call("command -v dhclient", expected_return_codes={0, 1}), call(f"dhclient -r {interface.name}", expected_return_codes={}), call(f"ip -4 addr flush dev {interface.name}"), ] @@ -195,6 +198,94 @@ def test_renew_ip(self, interface): ] ) + def test_release_ip_without_dhclient(self, interface): + interface._connection.execute_command.side_effect = [ + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ] + + interface.ip.release_ip(IPVersion.V4) + + interface._connection.execute_command.assert_has_calls( + [ + call("command -v dhclient", expected_return_codes={0, 1}), + call("command -v nmcli", expected_return_codes={0, 1}), + call("command -v wicked", expected_return_codes={0, 1}), + call(f"ip -4 addr flush dev {interface.name}"), + ] + ) + + def test_release_ip_with_wicked(self, interface): + interface._connection.execute_command.side_effect = [ + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="/usr/sbin/wicked", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ] + + interface.ip.release_ip(IPVersion.V4) + + interface._connection.execute_command.assert_has_calls( + [ + call("command -v dhclient", expected_return_codes={0, 1}), + call("command -v nmcli", expected_return_codes={0, 1}), + call("command -v wicked", expected_return_codes={0, 1}), + call(f"wicked ifreload {interface.name}", expected_return_codes={}), + call(f"ip -4 addr flush dev {interface.name}"), + ] + ) + + def test_release_ip_with_wicked_fallback(self, interface): + interface._connection.execute_command.side_effect = [ + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="/usr/sbin/wicked", stderr=""), + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr="reload failed"), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ] + + interface.ip.release_ip(IPVersion.V4) + + interface._connection.execute_command.assert_has_calls( + [ + call("command -v dhclient", expected_return_codes={0, 1}), + call("command -v nmcli", expected_return_codes={0, 1}), + call("command -v wicked", expected_return_codes={0, 1}), + call(f"wicked ifreload {interface.name}", expected_return_codes={}), + call(f"wicked ifdown --release --no-delete {interface.name}", expected_return_codes={}), + call(f"ip -4 addr flush dev {interface.name}"), + ] + ) + + def test_release_ip_with_nmcli_managed_dhcp(self, interface): + interface._connection.execute_command.side_effect = [ + ConnectionCompletedProcess(return_code=1, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="/usr/bin/nmcli", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="yes\n", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="br0\n", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="auto\nignore\n", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ConnectionCompletedProcess(return_code=0, args="", stdout="", stderr=""), + ] + + interface.ip.release_ip(IPVersion.V4) + + interface._connection.execute_command.assert_has_calls( + [ + call("command -v dhclient", expected_return_codes={0, 1}), + call("command -v nmcli", expected_return_codes={0, 1}), + call(f"nmcli -g GENERAL.NM-MANAGED device show {interface.name}", expected_return_codes={0, 1}), + call(f"nmcli -g GENERAL.CONNECTION device show {interface.name}", expected_return_codes={0, 1}), + call('nmcli -g ipv4.method,ipv6.method connection show "br0"', expected_return_codes={0, 1}), + call(f"nmcli device disconnect {interface.name}", expected_return_codes={}), + call(f"ip -4 addr flush dev {interface.name}"), + ] + ) + def test_get_dynamic_ip6_dhcp(self, interface): output = "2609135 ? Ss 0:00 dhclient -6 eth3" interface._connection.execute_command.return_value = ConnectionCompletedProcess(