diff --git a/blueman/main/DhcpClient.py b/blueman/main/DhcpClient.py index 3fd1a663b..a8c488918 100644 --- a/blueman/main/DhcpClient.py +++ b/blueman/main/DhcpClient.py @@ -28,6 +28,7 @@ def __init__(self, interface: str, timeout: int = 30) -> None: self._interface = interface self._timeout = timeout + self._client: "subprocess.Popen[bytes] | None" = None self._command = None for command in self.COMMANDS: @@ -50,12 +51,15 @@ def run(self) -> None: GLib.timeout_add(self._timeout * 1000, self._on_timeout) def _on_timeout(self) -> bool: - if not self._client.poll(): + if self._client is not None and self._client.poll() is None: logging.warning("Timeout reached, terminating DHCP client") self._client.terminate() return False def _check_client(self) -> bool: + if self._client is None: + return False + netifs = get_local_interfaces() status = self._client.poll() if status == 0: diff --git a/blueman/main/NetConf.py b/blueman/main/NetConf.py index 95813bc28..9b19f4f4f 100644 --- a/blueman/main/NetConf.py +++ b/blueman/main/NetConf.py @@ -88,7 +88,8 @@ def clean_up(self) -> None: pid = self._pid if pid is not None: - running_binary: str | None = next(binary for binary in self._BINARIES if _is_running(binary, pid)) + running_binary: str | None = next( + (binary for binary in self._BINARIES if _is_running(binary, pid)), None) if running_binary is not None: print('Terminating ' + running_binary) os.kill(pid, signal.SIGTERM) diff --git a/blueman/plugins/mechanism/Network.py b/blueman/plugins/mechanism/Network.py index b91813dca..9e5e5a35d 100644 --- a/blueman/plugins/mechanism/Network.py +++ b/blueman/plugins/mechanism/Network.py @@ -49,7 +49,10 @@ def dh_connected(_dh: DhcpClient, ip: str) -> None: def _enable_network(self, ip_address: str, netmask: str, dhcp_handler: str, address_changed: bool, caller: str) -> None: self.confirm_authorization(caller, "org.blueman.network.setup") - NetConf.apply_settings(ip_address, netmask, DHCPDHANDLERS[dhcp_handler], address_changed) + handler = DHCPDHANDLERS.get(dhcp_handler) + if handler is None: + raise ValueError(f"Unknown DHCP handler: {dhcp_handler!r}") + NetConf.apply_settings(ip_address, netmask, handler, address_changed) def _disable_network(self, caller: str) -> None: self.confirm_authorization(caller, "org.blueman.network.setup") diff --git a/blueman/plugins/mechanism/Rfcomm.py b/blueman/plugins/mechanism/Rfcomm.py index fc7ee17dd..84ae2908a 100644 --- a/blueman/plugins/mechanism/Rfcomm.py +++ b/blueman/plugins/mechanism/Rfcomm.py @@ -1,6 +1,7 @@ import os import subprocess import signal +import logging from blueman.Constants import RFCOMM_WATCHER_PATH from blueman.plugins.MechanismPlugin import MechanismPlugin @@ -11,7 +12,11 @@ def on_load(self) -> None: self.parent.add_method("CloseRFCOMM", ("n",), "", self._close_rfcomm) def _open_rfcomm(self, port_id: int) -> None: - subprocess.Popen([RFCOMM_WATCHER_PATH, f"/dev/rfcomm{port_id:d}"]) + try: + subprocess.Popen([RFCOMM_WATCHER_PATH, f"/dev/rfcomm{port_id:d}"]) + except OSError: + logging.exception(f"Failed to start rfcomm watcher for /dev/rfcomm{port_id:d}") + raise def _close_rfcomm(self, port_id: int) -> None: out, err = subprocess.Popen(['ps', '-e', 'o', 'pid,args'], stdout=subprocess.PIPE).communicate() diff --git a/test/main/test_dhcpclient.py b/test/main/test_dhcpclient.py new file mode 100644 index 000000000..0fc27458c --- /dev/null +++ b/test/main/test_dhcpclient.py @@ -0,0 +1,55 @@ +from unittest import TestCase +from unittest.mock import patch, Mock + +from blueman.main.DhcpClient import DhcpClient + + +class TestDhcpClientGuards(TestCase): + def _make(self): + # __init__ only probes for a client binary via have(); no process runs. + with patch("blueman.main.DhcpClient.have", return_value=None): + return DhcpClient("pan1") + + def test_client_initialised_to_none(self): + client = self._make() + self.assertIsNone(client._client) + + def test_check_client_before_run_does_not_crash(self): + # Previously self._client was undefined until run() -> AttributeError. + client = self._make() + self.assertFalse(client._check_client()) + + def test_on_timeout_before_run_does_not_crash(self): + client = self._make() + self.assertFalse(client._on_timeout()) + + def test_on_timeout_terminates_only_running_client(self): + # poll() == None -> still running -> terminate. + client = self._make() + client._client = Mock() + client._client.poll.return_value = None + client._on_timeout() + client._client.terminate.assert_called_once_with() + + def test_on_timeout_ignores_finished_client(self): + # poll() == 0 (success) or a non-zero exit code -> already done -> leave. + for status in (0, 1, 5, 255): + with self.subTest(status=status): + client = self._make() + client._client = Mock() + client._client.poll.return_value = status + client._on_timeout() + client._client.terminate.assert_not_called() + + def test_on_timeout_fuzz_poll_values(self): + for status in [None, 0, 1, -1, 2, 127, 255, 1000]: + with self.subTest(status=status): + client = self._make() + client._client = Mock() + client._client.poll.return_value = status + # Must never raise; terminate iff still running. + client._on_timeout() + if status is None: + client._client.terminate.assert_called_once_with() + else: + client._client.terminate.assert_not_called() diff --git a/test/main/test_netconf.py b/test/main/test_netconf.py index 19b1d0d6f..5c9cbfdcf 100644 --- a/test/main/test_netconf.py +++ b/test/main/test_netconf.py @@ -253,6 +253,54 @@ def test_cleanup(self, destroy_bridge_mock: Mock, call_mock: Mock, _create_bridg self.assertFalse(NetConf.locked("iptables")) +class TestDhcpHandlerCleanup(TestCase): + @patch("blueman.main.NetConf.os.kill") + @patch("blueman.main.NetConf.NetConf.unlock") + def test_no_running_binary_does_not_raise(self, unlock_mock: Mock, kill_mock: Mock) -> None: + # next() over an empty match must not raise StopIteration and abort + # the rest of clean_up(). + handler = DnsMasqHandler() + handler._pid = 1234 + with patch.object(NetConf, "locked", return_value=True), \ + patch("blueman.main.NetConf._is_running", return_value=False): + handler.clean_up() + kill_mock.assert_not_called() + unlock_mock.assert_called_once_with("dhcp") + + @patch("blueman.main.NetConf.os.kill") + @patch("blueman.main.NetConf.NetConf.unlock") + def test_running_binary_is_terminated(self, unlock_mock: Mock, kill_mock: Mock) -> None: + handler = DnsMasqHandler() + handler._pid = 4321 + with patch.object(NetConf, "locked", return_value=True), \ + patch("blueman.main.NetConf._is_running", return_value=True): + handler.clean_up() + kill_mock.assert_called_once() + self.assertEqual(kill_mock.call_args.args[0], 4321) + unlock_mock.assert_called_once_with("dhcp") + + @patch("blueman.main.NetConf.os.kill") + @patch("blueman.main.NetConf.NetConf.unlock") + def test_fuzz_is_running_combinations(self, unlock_mock: Mock, kill_mock: Mock) -> None: + # Multi-binary handler so the generator iterates more than once. + class MultiHandler(DnsMasqHandler): + _BINARIES = ["alpha", "beta", "gamma"] + + for pattern in [ + lambda b, p: False, + lambda b, p: True, + lambda b, p: b == "gamma", + lambda b, p: b in ("alpha", "beta"), + ]: + with self.subTest(pattern=pattern): + kill_mock.reset_mock() + handler = MultiHandler() + handler._pid = 777 + with patch.object(NetConf, "locked", return_value=True), \ + patch("blueman.main.NetConf._is_running", side_effect=pattern): + handler.clean_up() # must never raise + + class TestValidateIpv4(TestCase): def test_valid_addresses_pass(self) -> None: for addr, mask in [ diff --git a/test/plugins/mechanism/test_network.py b/test/plugins/mechanism/test_network.py new file mode 100644 index 000000000..4decacc0a --- /dev/null +++ b/test/plugins/mechanism/test_network.py @@ -0,0 +1,42 @@ +from unittest import TestCase +from unittest.mock import patch, Mock + +from blueman.plugins.mechanism import Network as network_module +from blueman.plugins.mechanism.Network import Network, DHCPDHANDLERS + + +def _make_network(): + # MechanismPlugin.__init__ takes timer/confirm_authorization off the parent + # and calls on_load(), which only registers methods on the (mock) parent. + return Network(Mock()) + + +class TestEnableNetworkHandlerValidation(TestCase): + def test_known_handlers_dispatch(self): + for name, cls in DHCPDHANDLERS.items(): + with self.subTest(handler=name): + net = _make_network() + with patch.object(network_module.NetConf, "apply_settings") as apply_mock: + net._enable_network("203.0.113.1", "255.255.255.0", name, False, ":1.1") + apply_mock.assert_called_once_with("203.0.113.1", "255.255.255.0", cls, False) + + def test_unknown_handler_raises_before_apply(self): + net = _make_network() + with patch.object(network_module.NetConf, "apply_settings") as apply_mock: + with self.assertRaises(ValueError): + net._enable_network("203.0.113.1", "255.255.255.0", "bogus", False, ":1.1") + apply_mock.assert_not_called() + + def test_fuzz_unknown_keys_never_keyerror(self): + bad_keys = [ + "", "dnsmasqhandler", "DnsMasqHandler ", " DnsMasqHandler", + "__class__", "../etc", "DnsMasq", "None", "0", "🚀", + "DnsMasqHandler\n", "DHCPDHANDLERS", + ] + for key in bad_keys: + with self.subTest(key=key): + net = _make_network() + with patch.object(network_module.NetConf, "apply_settings") as apply_mock: + with self.assertRaises(ValueError): + net._enable_network("203.0.113.1", "255.255.255.0", key, False, ":1.1") + apply_mock.assert_not_called() diff --git a/test/plugins/mechanism/test_rfcomm.py b/test/plugins/mechanism/test_rfcomm.py index 67ea46236..a56174e29 100644 --- a/test/plugins/mechanism/test_rfcomm.py +++ b/test/plugins/mechanism/test_rfcomm.py @@ -11,6 +11,31 @@ def _make_rfcomm(): return Rfcomm(Mock()) +class TestOpenRfcomm(TestCase): + def test_open_spawns_watcher(self): + rfcomm = _make_rfcomm() + with patch("blueman.plugins.mechanism.Rfcomm.subprocess.Popen") as popen_mock: + rfcomm._open_rfcomm(3) + args = popen_mock.call_args.args[0] + self.assertEqual(args[-1], "/dev/rfcomm3") + + def test_open_failure_logs_and_propagates(self): + rfcomm = _make_rfcomm() + with patch("blueman.plugins.mechanism.Rfcomm.subprocess.Popen", + side_effect=OSError("no such file")): + with self.assertLogs(level="ERROR"): + with self.assertRaises(OSError): + rfcomm._open_rfcomm(0) + + def test_open_fuzz_port_ids(self): + for port_id in [0, 1, 7, 15, 99, 12345]: + with self.subTest(port_id=port_id): + rfcomm = _make_rfcomm() + with patch("blueman.plugins.mechanism.Rfcomm.subprocess.Popen") as popen_mock: + rfcomm._open_rfcomm(port_id) + self.assertEqual(popen_mock.call_args.args[0][-1], f"/dev/rfcomm{port_id}") + + def _popen_returning(ps_output: str): proc = Mock() proc.communicate.return_value = (ps_output.encode("UTF-8"), None)