import hashlib import importlib import io import os import stat import subprocess import sys import tempfile import threading import time import types import unittest import zipfile from pathlib import Path from unittest.mock import Mock, patch from librenet_scanner import naabu_runtime from librenet_scanner.fastscan import BUNDLED_NAABU_PATH, SYSTEM_NAABU_CANDIDATES from librenet_scanner.privileged_helper import command_for ROOT = Path(__file__).resolve().parents[1] # Surface Qt minimale pour pouvoir exécuter réellement ScanWorker dans le builder. 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 NaabuProvisioning0422Tests(unittest.TestCase): def _archive(self, payload: bytes) -> bytes: stream = io.BytesIO() with zipfile.ZipFile(stream, "w", zipfile.ZIP_DEFLATED) as zf: zf.writestr("naabu", payload) return stream.getvalue() def test_runtime_constants_pin_official_261_amd64_release(self): self.assertEqual(naabu_runtime.NAABU_VERSION, "2.6.1") self.assertTrue(naabu_runtime.NAABU_AMD64_URL.endswith("/v2.6.1/naabu_2.6.1_linux_amd64.zip")) self.assertEqual( naabu_runtime.NAABU_AMD64_SHA256, "018c4c9884dea971eda860435ede3021d1150732f34cfd245498c6726d8cab90", ) self.assertEqual(BUNDLED_NAABU_PATH, "/usr/lib/librenet-scanner/bin/naabu") self.assertEqual(SYSTEM_NAABU_CANDIDATES[0], BUNDLED_NAABU_PATH) def test_offline_simulated_install_verifies_archive_and_installs_atomically(self): fake_binary = b"\x7fELF" + b"LibreNet fake Naabu test payload\n" archive = self._archive(fake_binary) sha = hashlib.sha256(archive).hexdigest() with tempfile.TemporaryDirectory() as td: destination = str(Path(td) / "bin" / "naabu") with patch.object(naabu_runtime, "_machine_is_amd64", return_value=True), \ patch.object(naabu_runtime, "naabu_version", return_value="2.6.1"): installed = naabu_runtime.install_naabu( destination=destination, url="https://example.invalid/naabu.zip", expected_sha256=sha, download_func=lambda _url: archive, require_root=False, ) self.assertEqual(installed, destination) self.assertEqual(Path(destination).read_bytes(), fake_binary) self.assertTrue(os.stat(destination).st_mode & stat.S_IXUSR) def test_checksum_mismatch_refuses_install(self): archive = self._archive(b"\x7fELFbad") with tempfile.TemporaryDirectory() as td: destination = str(Path(td) / "naabu") with patch.object(naabu_runtime, "_machine_is_amd64", return_value=True): with self.assertRaises(naabu_runtime.NaabuProvisionError): naabu_runtime.install_naabu( destination=destination, expected_sha256="0" * 64, download_func=lambda _url: archive, require_root=False, ) self.assertFalse(Path(destination).exists()) def test_version_probe_isolated_from_user_config_cloud_and_updates(self): completed = subprocess.CompletedProcess(args=[], returncode=0, stdout="Current Version: v2.6.1\n") with patch("librenet_scanner.naabu_runtime.subprocess.run", return_value=completed) as run: self.assertEqual(naabu_runtime.naabu_version("/tmp/naabu"), "2.6.1") args = run.call_args.args[0] self.assertIn("-disable-update-check", args) self.assertEqual(args[args.index("-config") + 1], "/dev/null") self.assertIn("-auth=false", args) def test_runtime_ignores_external_path_and_requires_private_exact_engine(self): fastscan = importlib.import_module("librenet_scanner.fastscan") with tempfile.TemporaryDirectory() as td: external = Path(td) / "naabu" external.write_text("#!/bin/sh\necho Current Version: v2.6.1\n") external.chmod(0o755) with patch.object(fastscan, "BUNDLED_NAABU_PATH", str(Path(td) / "missing-private-naabu")), \ patch.dict(os.environ, {"PATH": td}, clear=False): self.assertIsNone(fastscan.find_naabu()) def test_runtime_refuses_private_engine_with_wrong_version(self): fastscan = importlib.import_module("librenet_scanner.fastscan") with tempfile.TemporaryDirectory() as td: private = Path(td) / "naabu" private.write_text("fake") private.chmod(0o755) with patch.object(fastscan, "BUNDLED_NAABU_PATH", str(private)), \ patch.object(fastscan, "naabu_version", return_value="2.6.0"): self.assertIsNone(fastscan.find_naabu()) class NaabuPipeline0422Tests(unittest.TestCase): def test_standard_uses_discovery_then_fast_ports_in_user_or_admin_mode(self): for privileged in (False, True): request = scanner.ScanRequest( target="192.168.10.1", profile="Standard", interface=None, privileged=privileged ) worker = scanner.ScanWorker(request) worker._standard_baseline = Mock(return_value=({"192.168.10.1"}, "nmap")) worker._set_progress = Mock() worker._standard_scan() worker._standard_baseline.assert_called_once() def test_user_naabu_port_command_is_connect_pn_and_not_nmap(self): request = scanner.ScanRequest( target="192.168.10.1", profile="Standard", interface=None, privileged=False ) worker = scanner.ScanWorker(request) worker._run_naabu_json = Mock(return_value=[]) with patch("librenet_scanner.scanner.find_naabu", return_value=BUNDLED_NAABU_PATH): result = worker._naabu_ports(["192.168.10.1"], start_percent=20, end_percent=80) self.assertEqual(result, []) cmd = worker._run_naabu_json.call_args.args[0] self.assertEqual(cmd[0], BUNDLED_NAABU_PATH) self.assertEqual(cmd[cmd.index("-scan-type") + 1], "c") self.assertIn("-Pn", cmd) self.assertNotIn("nmap", " ".join(cmd).lower()) def test_nonprivileged_deep_command_can_be_stopped_while_silent(self): request = scanner.ScanRequest( target="192.168.10.1", profile="Approfondi", interface=None, privileged=False ) worker = scanner.ScanWorker(request) def request_stop(): time.sleep(0.25) worker.stop() threading.Thread(target=request_stop, daemon=True).start() started = time.monotonic() code, _out, _err = worker._run_command( [sys.executable, "-c", "import time; time.sleep(30)"], "Nmap approfondi silencieux", start_percent=1, end_percent=2, ) elapsed = time.monotonic() - started self.assertNotEqual(code, 0) self.assertLess(elapsed, 4.0, f"annulation Nmap trop lente: {elapsed:.2f}s") def test_admin_helper_prefers_librenet_runtime_and_keeps_port_prefilter_disabled(self): with patch("librenet_scanner.privileged_helper._trusted_binary", side_effect=lambda *candidates: candidates[0]) as trusted: cmd = command_for("naabu-standard", ["192.168.10.10"]) trusted.assert_called_once_with(BUNDLED_NAABU_PATH) self.assertEqual(cmd[0], BUNDLED_NAABU_PATH) self.assertIn("-Pn", cmd) self.assertNotIn("-wn", cmd) self.assertEqual(cmd[cmd.index("-scan-type") + 1], "s") self.assertIn("-stream", cmd) self.assertNotIn("-verify", cmd) self.assertNotIn("-retries", cmd) self.assertEqual(cmd[cmd.index("-warm-up-time") + 1], "0") def test_standard_profile_signature_changes_for_provisioned_runtime(self): self.assertIn("standard-v9:adaptive-nmap-small-naabu-large", scanner.PROFILE_SIGNATURES["Standard"]) self.assertIn("deep-v10:adaptive-standard-baseline", scanner.PROFILE_SIGNATURES["Approfondi"]) def test_silent_naabu_can_be_stopped_without_waiting_for_stdout(self): request = scanner.ScanRequest( target="192.168.10.1", profile="Standard", interface=None, privileged=False ) worker = scanner.ScanWorker(request) def request_stop(): time.sleep(0.25) worker.stop() stopper = threading.Thread(target=request_stop, daemon=True) stopper.start() started = time.monotonic() result = worker._run_naabu_json( [sys.executable, "-c", "import time; time.sleep(30)"], "test Naabu silencieux", lambda _line: None, start_percent=1, end_percent=2, phase="le scan de ports", ) elapsed = time.monotonic() - started stopper.join(timeout=1) self.assertEqual(result, []) self.assertLess(elapsed, 4.0, f"annulation trop lente: {elapsed:.2f}s") def test_detailed_worker_stop_terminates_its_process_group(self): worker = scanner.HostScanWorker("127.0.0.1", privileged=False) proc = subprocess.Popen( [sys.executable, "-c", "import time; time.sleep(30)"], start_new_session=True, ) worker._proc = proc started = time.monotonic() worker.stop() proc.wait(timeout=4) self.assertLess(time.monotonic() - started, 4.0) self.assertTrue(worker.isInterruptionRequested()) class Packaging0422Tests(unittest.TestCase): def test_debian_metadata_is_amd64_and_installation_has_no_mandatory_network_access(self): control = (ROOT / "packaging/debian/control").read_text(encoding="utf-8") postinst = (ROOT / "packaging/debian/postinst").read_text(encoding="utf-8") build = (ROOT / "packaging/build-deb.sh").read_text(encoding="utf-8") self.assertIn("Version: 1.0.0", control) self.assertIn("Architecture: amd64", control) self.assertIn("set -eu", postinst) self.assertNotIn("librenet-scanner-install-naabu --ensure", postinst) self.assertIn("Aucune dépendance réseau", postinst) self.assertIn("VERSION=1.0.0", build) self.assertIn("ARCH=amd64", build) def test_runtime_installer_is_packaged_and_postrm_removes_runtime(self): build = (ROOT / "packaging/build-deb.sh").read_text(encoding="utf-8") postrm = (ROOT / "packaging/debian/postrm").read_text(encoding="utf-8") self.assertIn("librenet-scanner-install-naabu", build) self.assertIn("/usr/lib/librenet-scanner/bin", build) self.assertIn("NAABU-LICENSE.txt", build) self.assertIn("/usr/lib/librenet-scanner/bin/naabu", postrm) if __name__ == "__main__": unittest.main()