import importlib import sys import threading import time import types import unittest from unittest.mock import Mock, patch if "PySide6.QtCore" not in sys.modules: qtcore = types.ModuleType("PySide6.QtCore") pyside = types.ModuleType("PySide6") class DummySignal: def __init__(self, *args, **kwargs): self.calls = [] def emit(self, *args): self.calls.append(args) class DummyQThread: def __init__(self, parent=None): self._interrupted = False def requestInterruption(self): self._interrupted = True def isInterruptionRequested(self): return self._interrupted qtcore.Signal = DummySignal qtcore.QThread = DummyQThread pyside.QtCore = qtcore sys.modules["PySide6"] = pyside sys.modules["PySide6.QtCore"] = qtcore scanner = importlib.import_module("librenet_scanner.scanner") class PerformancePipeline0423Tests(unittest.TestCase): def worker(self, privileged=False, target="192.168.10.0/24"): return scanner.ScanWorker(scanner.ScanRequest(target, "Standard", None, privileged)) def test_user_case_four_live_hosts_uses_bounded_nmap_not_naabu(self): worker = self.worker() live = {"192.168.10.1", "192.168.10.22", "192.168.10.226", "192.168.10.254"} worker._standard_discovery = Mock(return_value=live) worker._naabu_ports = Mock(side_effect=AssertionError("Naabu ne doit pas démarrer pour 4 hôtes")) worker._nmap_standard_ports = Mock(return_value=[]) worker._neighbor_hosts = Mock(return_value=[]) worker._set_progress = Mock() result, engine = worker._standard_baseline(start_percent=4, end_percent=96) self.assertEqual(result, live) self.assertEqual(engine, "nmap") worker._naabu_ports.assert_not_called() scanned = worker._nmap_standard_ports.call_args.args[0] self.assertEqual(scanned, sorted(live, key=scanner.ipaddress.ip_address)) self.assertFalse(worker._nmap_standard_ports.call_args.kwargs["fallback"]) def test_large_active_set_uses_naabu_only_on_discovered_hosts(self): worker = self.worker() live = {f"192.168.10.{i}" for i in range(1, scanner.NAABU_ACTIVE_HOST_THRESHOLD + 1)} worker._standard_discovery = Mock(return_value=live) worker._naabu_ports = Mock(return_value=[]) worker._nmap_standard_ports = Mock(side_effect=AssertionError("pas de Nmap ports nominal sur grand ensemble")) worker._neighbor_hosts = Mock(return_value=[]) result, engine = worker._standard_baseline(start_percent=4, end_percent=96) self.assertEqual(result, live) self.assertEqual(engine, "naabu") scanned = worker._naabu_ports.call_args.args[0] self.assertEqual(scanned, sorted(live, key=scanner.ipaddress.ip_address)) self.assertEqual(len(scanned), scanner.NAABU_ACTIVE_HOST_THRESHOLD) def test_naabu_is_split_into_small_bounded_batches(self): worker = self.worker() ips = [f"192.168.10.{i}" for i in range(1, 71)] worker._run_naabu_json = Mock(return_value=[]) with patch("librenet_scanner.scanner.find_naabu", return_value="/usr/lib/librenet-scanner/bin/naabu"): result = worker._naabu_ports(ips, start_percent=20, end_percent=80) self.assertEqual(result, []) self.assertEqual(worker._run_naabu_json.call_count, 3) for call in worker._run_naabu_json.call_args_list: cmd = call.args[0] hosts = cmd[cmd.index("-host") + 1].split(",") self.assertLessEqual(len(hosts), scanner.NAABU_BATCH_SIZE) self.assertIn("-stream", cmd) self.assertNotIn("-verify", cmd) self.assertNotIn("-retries", cmd) self.assertEqual(call.kwargs["timeout_seconds"], scanner.NAABU_BATCH_TIMEOUT_SECONDS) def test_silent_naabu_batch_hits_wall_clock_timeout(self): worker = self.worker(target="127.0.0.1") started = time.monotonic() result = worker._run_naabu_json( [sys.executable, "-c", "import time; time.sleep(30)"], "Naabu timeout test", lambda _line: None, start_percent=1, end_percent=2, phase="le scan de ports", timeout_seconds=0.30, ) elapsed = time.monotonic() - started self.assertIsNone(result) self.assertFalse(worker.isInterruptionRequested()) self.assertLess(elapsed, 4.0) def test_generic_command_timeout_is_bounded(self): worker = self.worker(target="127.0.0.1") started = time.monotonic() code, _out, err = worker._run_command( [sys.executable, "-c", "import time; time.sleep(30)"], "timeout test", timeout_seconds=0.30, ) elapsed = time.monotonic() - started self.assertEqual(code, 124) self.assertIn("Délai maximal dépassé", err) self.assertLess(elapsed, 4.0) def test_arp_scan_itself_is_bounded(self): worker = self.worker() iface = types.SimpleNamespace( name="enp42s0", address="192.168.10.1", prefixlen=24, network="192.168.10.0/24", mac="", ) worker.request.interface = iface worker._run_command = Mock(return_value=(0, "", "")) with patch("librenet_scanner.scanner.find_arp_scan", return_value="/usr/bin/arp-scan"): worker._emit_arp(start_percent=5, end_percent=15) self.assertEqual(worker._run_command.call_args.kwargs["timeout_seconds"], 8.0) def test_standard_discovery_is_single_fast_nmap_pass_after_arp(self): worker = self.worker() worker._emit_local_host = Mock(return_value=[]) worker._emit_arp = Mock(return_value=[]) worker._neighbor_hosts = Mock(return_value=[]) worker._set_progress = Mock() worker._nmap = Mock(return_value=[]) result = worker._standard_discovery( start_percent=4, end_percent=35, target_ips=[f"192.168.10.{i}" for i in range(1, 255)], ) self.assertEqual(result, set()) worker._nmap.assert_called_once() cmd = worker._nmap.call_args.args[0] self.assertEqual(cmd[:6], ["nmap", "-sn", "-n", "-T4", "--max-retries", "1"]) self.assertIsNotNone(worker._nmap.call_args.kwargs["timeout_seconds"]) def test_naabu_failure_does_not_rescan_dead_addresses_with_nmap(self): worker = self.worker() live = {f"192.168.10.{i}" for i in range(1, scanner.NAABU_ACTIVE_HOST_THRESHOLD + 1)} worker._standard_discovery = Mock(return_value=live) worker._naabu_ports = Mock(return_value=None) worker._nmap_standard_ports = Mock(return_value=[]) worker._neighbor_hosts = Mock(return_value=[]) worker._set_progress = Mock() _known, engine = worker._standard_baseline(start_percent=4, end_percent=96) self.assertEqual(engine, "nmap-fallback") self.assertEqual(worker._nmap_standard_ports.call_args.args[0], sorted(live, key=scanner.ipaddress.ip_address)) self.assertTrue(worker._nmap_standard_ports.call_args.kwargs["fallback"]) def test_small_nmap_profile_is_bounded_and_only_receives_live_hosts(self): worker = self.worker() worker._nmap = Mock(return_value=[]) live = ["192.168.10.22", "192.168.10.226"] result = worker._nmap_standard_ports(live, start_percent=20, end_percent=80, fallback=False) self.assertEqual(result, []) cmd = worker._nmap.call_args.args[0] self.assertEqual(cmd[-2:], live) self.assertIn("-n", cmd) self.assertEqual(cmd[cmd.index("--max-retries") + 1], "1") self.assertEqual(cmd[cmd.index("--host-timeout") + 1], "12s") self.assertIsNotNone(worker._nmap.call_args.kwargs["timeout_seconds"]) def test_privileged_small_nmap_profile_disables_dns_and_bounds_hosts(self): with patch("librenet_scanner.privileged_helper._trusted_binary", return_value="/usr/bin/nmap"): from librenet_scanner.privileged_helper import command_for cmd = command_for("nmap-standard", ["192.168.10.22", "192.168.10.226"]) self.assertIn("-n", cmd) self.assertEqual(cmd[cmd.index("--max-retries") + 1], "1") self.assertEqual(cmd[cmd.index("--host-timeout") + 1], "12s") self.assertEqual(cmd[-2:], ["192.168.10.22", "192.168.10.226"]) def test_admin_naabu_command_is_fast_profile(self): with patch("librenet_scanner.privileged_helper._trusted_binary", return_value="/usr/lib/librenet-scanner/bin/naabu"): from librenet_scanner.privileged_helper import command_for cmd = command_for("naabu-standard", ["192.168.10.22"]) self.assertIn("-stream", cmd) self.assertEqual(cmd[cmd.index("-c") + 1], "100") self.assertEqual(cmd[cmd.index("-rate") + 1], "2500") self.assertEqual(cmd[cmd.index("-timeout") + 1], "800ms") self.assertNotIn("-verify", cmd) self.assertNotIn("-retries", cmd) if __name__ == "__main__": unittest.main()