From aed1cd7950eae623fa564398a5b13939c2fe29f4 Mon Sep 17 00:00:00 2001 From: wind Date: Fri, 27 Feb 2026 19:54:09 +0100 Subject: [PATCH 1/4] feat: add arp-scan/nmap integration and XML parsing (#4) - Implement ArpHost dataclass and parse_arp_output() pure parser - Implement run_arp_scan() with subprocess, env var fallbacks, and error handling - Fix arp-scan regex to handle optional vendor column (no trailing tab) - Implement NmapHost/PortInfo dataclasses and parse_nmap_xml() with full XML parsing - Implement run_nmap_scan() with temp-file XML output and guaranteed cleanup - Implement orchestrate_scan() merging ARP MAC enrichment into nmap results - Add 4 nmap fixture XML files covering rich, minimal, down-host, and no-ports cases - Add 45 tests (unit + integration) with 100% line coverage - Fix pre-existing Ruff PT001/PT023 issues in test_db.py and test_main.py --- backend/app/scanner/__init__.py | 71 ++- backend/app/scanner/arp_scan.py | 68 +++ backend/app/scanner/nmap_scan.py | 191 ++++++++ backend/tests/fixtures/nmap_host_down.xml | 13 + backend/tests/fixtures/nmap_minimal_host.xml | 19 + .../tests/fixtures/nmap_no_ports_element.xml | 13 + backend/tests/fixtures/nmap_two_hosts.xml | 52 +++ backend/tests/test_db.py | 18 +- backend/tests/test_main.py | 8 +- backend/tests/test_scanner.py | 416 ++++++++++++++++++ 10 files changed, 855 insertions(+), 14 deletions(-) create mode 100644 backend/app/scanner/arp_scan.py create mode 100644 backend/app/scanner/nmap_scan.py create mode 100644 backend/tests/fixtures/nmap_host_down.xml create mode 100644 backend/tests/fixtures/nmap_minimal_host.xml create mode 100644 backend/tests/fixtures/nmap_no_ports_element.xml create mode 100644 backend/tests/fixtures/nmap_two_hosts.xml create mode 100644 backend/tests/test_scanner.py diff --git a/backend/app/scanner/__init__.py b/backend/app/scanner/__init__.py index 9305fef..fc2e3c4 100644 --- a/backend/app/scanner/__init__.py +++ b/backend/app/scanner/__init__.py @@ -1 +1,70 @@ -"""Scanner package — nmap + arp-scan integration.""" +""" +scanner/__init__.py — Public interface for the scanner package. + +The orchestrate_scan() function is the single entry point called by the +scheduler and the manual-trigger API endpoint. +""" + +from __future__ import annotations + +import logging +import os +from dataclasses import dataclass, field + +from app.scanner.arp_scan import ArpHost, run_arp_scan +from app.scanner.nmap_scan import NmapHost, run_nmap_scan + +logger = logging.getLogger(__name__) + + +@dataclass +class ScanResult: + """Merged result of one full LAN scan cycle.""" + + hosts: list[NmapHost] = field(default_factory=list) + """nmap-enriched hosts (ip, mac, hostname, os_guess, ports).""" + + arp_only: list[ArpHost] = field(default_factory=list) + """Hosts found by arp-scan that nmap returned no data for.""" + + +def orchestrate_scan( + interface: str | None = None, + subnet: str | None = None, +) -> ScanResult: + """ + Run a full scan cycle: + 1. arp-scan → enumerate live IPs + 2. nmap → service/version/OS detection on those IPs + 3. Merge → attach MAC/vendor from arp to nmap results + + Returns a ScanResult. Raises RuntimeError on tool failure. + """ + iface = interface or os.environ.get("NETWORK_INTERFACE", "eth0") + net = subnet or os.environ.get("SCAN_SUBNET", "192.168.1.0/24") + + logger.info("Starting ARP scan on %s / %s", iface, net) + arp_hosts = run_arp_scan(interface=iface, subnet=net) + logger.info("ARP scan found %d hosts", len(arp_hosts)) + + if not arp_hosts: + return ScanResult() + + ip_list = [h.ip for h in arp_hosts] + arp_by_ip = {h.ip: h for h in arp_hosts} + + logger.info("Starting nmap scan on %d hosts", len(ip_list)) + nmap_hosts = run_nmap_scan(hosts=ip_list, interface=iface) + logger.info("nmap scan returned %d hosts", len(nmap_hosts)) + + # Enrich nmap results with MAC / vendor from arp-scan + nmap_ips: set[str] = set() + for nh in nmap_hosts: + nmap_ips.add(nh.ip) + if not nh.mac and nh.ip in arp_by_ip: + nh.mac = arp_by_ip[nh.ip].mac + + # Hosts arp found but nmap returned nothing for (e.g. ICMP-filtered) + arp_only = [h for h in arp_hosts if h.ip not in nmap_ips] + + return ScanResult(hosts=nmap_hosts, arp_only=arp_only) diff --git a/backend/app/scanner/arp_scan.py b/backend/app/scanner/arp_scan.py new file mode 100644 index 0000000..8bfce9f --- /dev/null +++ b/backend/app/scanner/arp_scan.py @@ -0,0 +1,68 @@ +""" +arp_scan.py — Invoke arp-scan to enumerate live hosts on the LAN. + +All subprocess calls go through run_arp_scan(); the rest of the module is +pure parsing so it can be unit-tested without a real network. +""" + +from __future__ import annotations + +import logging +import os +import re +import subprocess +from dataclasses import dataclass + +logger = logging.getLogger(__name__) + +# Matches a line like: 192.168.1.1\taa:bb:cc:dd:ee:ff\tVendor Name +# Vendor column is optional — arp-scan omits the tab when the vendor is unknown. +_ARP_LINE_RE = re.compile( + r"^(?P\d{1,3}(?:\.\d{1,3}){3})\t(?P[0-9a-fA-F:]{17})(?:\t(?P.*))?$" +) + + +@dataclass +class ArpHost: + ip: str + mac: str + vendor: str = "" + + +def run_arp_scan( + interface: str | None = None, + subnet: str | None = None, +) -> list[ArpHost]: + """ + Run arp-scan and return a list of discovered hosts. + + Falls back to env vars NETWORK_INTERFACE / SCAN_SUBNET when arguments + are not provided. Raises RuntimeError if arp-scan exits non-zero. + """ + iface = interface or os.environ.get("NETWORK_INTERFACE", "eth0") + target = subnet or os.environ.get("SCAN_SUBNET", "192.168.1.0/24") + + cmd = ["arp-scan", "--interface", iface, "--localnet", target] + logger.debug("arp-scan command: %s", cmd) + + result = subprocess.run( # noqa: S603 — argv list, no shell injection + cmd, + capture_output=True, + text=True, + timeout=60, + ) + + if result.returncode not in (0, 1): # arp-scan returns 1 when no hosts found + raise RuntimeError(f"arp-scan failed (exit {result.returncode}): {result.stderr.strip()}") + + return parse_arp_output(result.stdout) + + +def parse_arp_output(output: str) -> list[ArpHost]: + """Parse raw arp-scan stdout into ArpHost objects.""" + hosts: list[ArpHost] = [] + for line in output.splitlines(): + m = _ARP_LINE_RE.match(line.strip()) + if m: + hosts.append(ArpHost(ip=m["ip"], mac=m["mac"], vendor=(m["vendor"] or "").strip())) + return hosts diff --git a/backend/app/scanner/nmap_scan.py b/backend/app/scanner/nmap_scan.py new file mode 100644 index 0000000..81dcb10 --- /dev/null +++ b/backend/app/scanner/nmap_scan.py @@ -0,0 +1,191 @@ +""" +nmap_scan.py — Invoke nmap for service/version/OS detection and parse XML output. + +run_nmap_scan() is the only function that touches subprocess; everything else +is pure XML parsing so it can be integration-tested with fixture files. +""" + +from __future__ import annotations + +import logging +import os +import subprocess +import tempfile +import xml.etree.ElementTree as ET +from dataclasses import dataclass, field +from pathlib import Path + +logger = logging.getLogger(__name__) + + +# ── Data classes ───────────────────────────────────────────────────────────── + + +@dataclass +class PortInfo: + port_number: int + protocol: str + state: str + service_name: str = "" + version_banner: str = "" + + +@dataclass +class NmapHost: + ip: str + mac: str = "" + hostname: str = "" + os_guess: str = "" + ports: list[PortInfo] = field(default_factory=list) + + +# ── Public API ──────────────────────────────────────────────────────────────── + + +def run_nmap_scan( + hosts: list[str], + interface: str | None = None, +) -> list[NmapHost]: + """ + Run nmap -sV -O --top-ports 1000 -oX against *hosts* and return parsed results. + + Uses a temporary file for XML output so the subprocess and parser are + independently testable. Raises RuntimeError on non-zero exit. + """ + if not hosts: + return [] + + iface = interface or os.environ.get("NETWORK_INTERFACE", "eth0") + + with tempfile.NamedTemporaryFile(suffix=".xml", delete=False) as tmp: + xml_path = tmp.name + + try: + cmd = [ + "nmap", + "-sV", # service/version detection + "-O", # OS detection (best-effort) + "--top-ports", + "1000", + "-e", + iface, + "-oX", + xml_path, + *hosts, + ] + logger.debug("nmap command: %s", cmd) + + result = subprocess.run( # noqa: S603 — argv list, no shell injection + cmd, + capture_output=True, + text=True, + timeout=300, + ) + + if result.returncode != 0: + raise RuntimeError(f"nmap failed (exit {result.returncode}): {result.stderr.strip()}") + + return parse_nmap_xml(Path(xml_path).read_text()) + finally: + Path(xml_path).unlink(missing_ok=True) + + +def parse_nmap_xml(xml_text: str) -> list[NmapHost]: + """ + Parse nmap XML output (``-oX``) into a list of NmapHost objects. + + Only 'up' hosts with at least one port entry are included; filtered + ports (state != 'open') are silently skipped. + """ + root = ET.fromstring(xml_text) # noqa: S314 — input is local nmap output, not user data + results: list[NmapHost] = [] + + for host_el in root.findall("host"): + status = host_el.find("status") + if status is None or status.get("state") != "up": + continue + + host = _parse_host(host_el) + results.append(host) + + return results + + +# ── Private helpers ─────────────────────────────────────────────────────────── + + +def _parse_host(host_el: ET.Element) -> NmapHost: + ip = "" + mac = "" + + for addr in host_el.findall("address"): + atype = addr.get("addrtype", "") + if atype == "ipv4": + ip = addr.get("addr", "") + elif atype == "mac": + mac = addr.get("addr", "") + + hostname = _parse_hostname(host_el) + os_guess = _parse_os(host_el) + ports = _parse_ports(host_el) + + return NmapHost(ip=ip, mac=mac, hostname=hostname, os_guess=os_guess, ports=ports) + + +def _parse_hostname(host_el: ET.Element) -> str: + hostnames = host_el.find("hostnames") + if hostnames is None: + return "" + for hn in hostnames.findall("hostname"): + name = hn.get("name", "") + if name: + return name + return "" + + +def _parse_os(host_el: ET.Element) -> str: + os_el = host_el.find("os") + if os_el is None: + return "" + # Prefer the match with the highest accuracy + best: tuple[int, str] = (0, "") + for match in os_el.findall("osmatch"): + accuracy = int(match.get("accuracy", "0")) + name = match.get("name", "") + if accuracy > best[0]: + best = (accuracy, name) + return best[1] + + +def _parse_ports(host_el: ET.Element) -> list[PortInfo]: + ports_el = host_el.find("ports") + if ports_el is None: + return [] + + ports: list[PortInfo] = [] + for port_el in ports_el.findall("port"): + state_el = port_el.find("state") + if state_el is None or state_el.get("state") != "open": + continue + + service_el = port_el.find("service") + service_name = "" + version_banner = "" + if service_el is not None: + service_name = service_el.get("name", "") + product = service_el.get("product", "") + version = service_el.get("version", "") + extra = service_el.get("extrainfo", "") + parts = filter(None, [product, version, extra]) + version_banner = " ".join(parts) + + ports.append( + PortInfo( + port_number=int(port_el.get("portid", "0")), + protocol=port_el.get("protocol", "tcp"), + state=state_el.get("state", ""), + service_name=service_name, + version_banner=version_banner, + ) + ) + return ports diff --git a/backend/tests/fixtures/nmap_host_down.xml b/backend/tests/fixtures/nmap_host_down.xml new file mode 100644 index 0000000..8b7b397 --- /dev/null +++ b/backend/tests/fixtures/nmap_host_down.xml @@ -0,0 +1,13 @@ + + + + + + + +
+ + + + diff --git a/backend/tests/fixtures/nmap_minimal_host.xml b/backend/tests/fixtures/nmap_minimal_host.xml new file mode 100644 index 0000000..cc778ae --- /dev/null +++ b/backend/tests/fixtures/nmap_minimal_host.xml @@ -0,0 +1,19 @@ + + + + + + + +
+
+ + + + + + + + + diff --git a/backend/tests/fixtures/nmap_no_ports_element.xml b/backend/tests/fixtures/nmap_no_ports_element.xml new file mode 100644 index 0000000..7b27da9 --- /dev/null +++ b/backend/tests/fixtures/nmap_no_ports_element.xml @@ -0,0 +1,13 @@ + + + + + + + +
+
+ + + diff --git a/backend/tests/fixtures/nmap_two_hosts.xml b/backend/tests/fixtures/nmap_two_hosts.xml new file mode 100644 index 0000000..da4e3fa --- /dev/null +++ b/backend/tests/fixtures/nmap_two_hosts.xml @@ -0,0 +1,52 @@ + + + + + + + +
+
+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +
+ + + + + + + + + + + diff --git a/backend/tests/test_db.py b/backend/tests/test_db.py index 4ce59ab..b147007 100644 --- a/backend/tests/test_db.py +++ b/backend/tests/test_db.py @@ -9,7 +9,7 @@ from sqlalchemy.orm import sessionmaker -@pytest.fixture() +@pytest.fixture def in_memory_engine(): """Provide a fresh in-memory SQLite engine with tables created.""" import app.models.device # noqa: F401 — register ORM models @@ -24,19 +24,19 @@ def in_memory_engine(): Base.metadata.drop_all(bind=engine) -@pytest.mark.unit() +@pytest.mark.unit def test_init_db_creates_devices_table(in_memory_engine): inspector = inspect(in_memory_engine) assert "devices" in inspector.get_table_names() -@pytest.mark.unit() +@pytest.mark.unit def test_init_db_creates_ports_table(in_memory_engine): inspector = inspect(in_memory_engine) assert "ports" in inspector.get_table_names() -@pytest.mark.unit() +@pytest.mark.unit def test_devices_table_columns(in_memory_engine): inspector = inspect(in_memory_engine) cols = {c["name"] for c in inspector.get_columns("devices")} @@ -51,14 +51,14 @@ def test_devices_table_columns(in_memory_engine): } <= cols -@pytest.mark.unit() +@pytest.mark.unit def test_ports_table_columns(in_memory_engine): inspector = inspect(in_memory_engine) cols = {c["name"] for c in inspector.get_columns("ports")} assert {"id", "device_id", "port_number", "protocol", "service_name", "version_banner"} <= cols -@pytest.mark.integration() +@pytest.mark.integration def test_get_db_yields_and_closes(in_memory_engine, monkeypatch): """get_db dependency yields a session and closes it after iteration.""" from app import db as db_module @@ -76,7 +76,7 @@ def test_get_db_yields_and_closes(in_memory_engine, monkeypatch): pass -@pytest.mark.integration() +@pytest.mark.integration def test_device_crud(in_memory_engine): """Basic Device create/read round-trip against in-memory DB.""" from app.models.device import Device @@ -93,7 +93,7 @@ def test_device_crud(in_memory_engine): assert fetched.hostname == "router" -@pytest.mark.integration() +@pytest.mark.integration def test_port_crud_with_device(in_memory_engine): """Port linked to Device creates FK relationship correctly.""" from app.models.device import Device, Port @@ -113,7 +113,7 @@ def test_port_crud_with_device(in_memory_engine): assert fetched_port.device_id == device.id -@pytest.mark.integration() +@pytest.mark.integration def test_device_cascade_deletes_ports(in_memory_engine): """Deleting a Device cascades to its Ports.""" from app.models.device import Device, Port diff --git a/backend/tests/test_main.py b/backend/tests/test_main.py index 0500af2..e5c1733 100644 --- a/backend/tests/test_main.py +++ b/backend/tests/test_main.py @@ -20,19 +20,19 @@ def client(): yield c -@pytest.mark.unit() +@pytest.mark.unit def test_health_returns_200(client): response = client.get("/health") assert response.status_code == 200 -@pytest.mark.unit() +@pytest.mark.unit def test_health_body(client): response = client.get("/health") assert response.json() == {"status": "ok"} -@pytest.mark.unit() +@pytest.mark.unit def test_openapi_schema_accessible(client): response = client.get("/openapi.json") assert response.status_code == 200 @@ -40,7 +40,7 @@ def test_openapi_schema_accessible(client): assert data["info"]["title"] == "NetworkCrawler" -@pytest.mark.unit() +@pytest.mark.unit def test_docs_accessible(client): response = client.get("/docs") assert response.status_code == 200 diff --git a/backend/tests/test_scanner.py b/backend/tests/test_scanner.py new file mode 100644 index 0000000..b9f5303 --- /dev/null +++ b/backend/tests/test_scanner.py @@ -0,0 +1,416 @@ +""" +test_scanner.py — Tests for the scanner package. + +Unit tests mock subprocess so no real network tools are required. +Integration tests use fixture XML files to exercise the XML parsers end-to-end. + +Markers +------- +unit — pure logic / mocked I/O, always run +integration — fixture-based, no live network +""" + +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from app.scanner import ScanResult, orchestrate_scan +from app.scanner.arp_scan import ArpHost, parse_arp_output, run_arp_scan +from app.scanner.nmap_scan import NmapHost, parse_nmap_xml, run_nmap_scan + +# ── Fixture helpers ────────────────────────────────────────────────────────── + +FIXTURES = Path(__file__).parent / "fixtures" + + +def _fixture(name: str) -> str: + return (FIXTURES / name).read_text() + + +# ═══════════════════════════════════════════════════════════════════════════════ +# arp_scan — unit tests +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestParseArpOutput: + """parse_arp_output() is pure; no mocking needed.""" + + @pytest.mark.unit + def test_parses_single_host(self): + output = "192.168.1.1\taa:bb:cc:dd:ee:ff\tCisco Systems\n" + hosts = parse_arp_output(output) + assert len(hosts) == 1 + assert hosts[0].ip == "192.168.1.1" + assert hosts[0].mac == "aa:bb:cc:dd:ee:ff" + assert hosts[0].vendor == "Cisco Systems" + + @pytest.mark.unit + def test_parses_multiple_hosts(self): + output = ( + "192.168.1.1\taa:bb:cc:dd:ee:01\tVendorA\n192.168.1.2\taa:bb:cc:dd:ee:02\tVendorB\n" + ) + hosts = parse_arp_output(output) + assert len(hosts) == 2 + assert hosts[1].ip == "192.168.1.2" + + @pytest.mark.unit + def test_skips_header_and_summary_lines(self): + output = ( + "Interface: eth0, datalink type: EN10MB (Ethernet)\n" + "Starting arp-scan 1.10.0 with 256 hosts...\n" + "192.168.1.1\taa:bb:cc:dd:ee:ff\tCisco Systems\n" + "\n" + "3 packets received by filter, 0 packets dropped by kernel\n" + ) + hosts = parse_arp_output(output) + assert len(hosts) == 1 + + @pytest.mark.unit + def test_empty_output_returns_empty_list(self): + assert parse_arp_output("") == [] + + @pytest.mark.unit + def test_vendor_can_be_blank_with_empty_tab(self): + """Trailing tab present but vendor field is empty.""" + output = "10.0.0.5\t00:11:22:33:44:55\t\n" + hosts = parse_arp_output(output) + assert len(hosts) == 1 + assert hosts[0].vendor == "" + + @pytest.mark.unit + def test_vendor_can_be_blank_no_tab(self): + """arp-scan omits the vendor tab entirely when vendor is unknown.""" + output = "10.0.0.5\t00:11:22:33:44:55\n" + hosts = parse_arp_output(output) + assert len(hosts) == 1 + assert hosts[0].vendor == "" + + @pytest.mark.unit + def test_strips_whitespace_around_lines(self): + output = " 192.168.1.1\tde:ad:be:ef:00:01\tSome Corp \n" + hosts = parse_arp_output(output) + assert hosts[0].ip == "192.168.1.1" + + +class TestRunArpScan: + """run_arp_scan() mocks subprocess.run.""" + + @pytest.mark.unit + def test_returns_parsed_hosts_on_success(self): + stdout = "192.168.1.1\taa:bb:cc:dd:ee:ff\tCisco\n" + mock_result = MagicMock(returncode=0, stdout=stdout, stderr="") + + with patch("app.scanner.arp_scan.subprocess.run", return_value=mock_result) as mock_run: + hosts = run_arp_scan(interface="eth0", subnet="192.168.1.0/24") + + assert len(hosts) == 1 + assert hosts[0].ip == "192.168.1.1" + # Verify correct command was built + cmd = mock_run.call_args[0][0] + assert "arp-scan" in cmd + assert "--interface" in cmd + assert "eth0" in cmd + + @pytest.mark.unit + def test_returns_empty_list_when_no_hosts_found(self): + """arp-scan exits 1 when no hosts are found — must not raise.""" + mock_result = MagicMock(returncode=1, stdout="", stderr="") + with patch("app.scanner.arp_scan.subprocess.run", return_value=mock_result): + hosts = run_arp_scan(interface="eth0", subnet="192.168.1.0/24") + assert hosts == [] + + @pytest.mark.unit + def test_raises_on_nonzero_exit(self): + mock_result = MagicMock(returncode=2, stdout="", stderr="permission denied") + with patch("app.scanner.arp_scan.subprocess.run", return_value=mock_result): + with pytest.raises(RuntimeError, match="arp-scan failed"): + run_arp_scan() + + @pytest.mark.unit + def test_uses_env_vars_as_defaults(self, monkeypatch): + monkeypatch.setenv("NETWORK_INTERFACE", "ens3") + monkeypatch.setenv("SCAN_SUBNET", "10.0.0.0/8") + mock_result = MagicMock(returncode=0, stdout="", stderr="") + with patch("app.scanner.arp_scan.subprocess.run", return_value=mock_result) as mock_run: + run_arp_scan() + cmd = mock_run.call_args[0][0] + assert "ens3" in cmd + assert "10.0.0.0/8" in cmd + + +# ═══════════════════════════════════════════════════════════════════════════════ +# nmap_scan — integration tests (fixture XML) +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestParseNmapXml: + """parse_nmap_xml() is pure; exercises fixture files.""" + + @pytest.mark.integration + def test_two_hosts_fixture(self): + xml = _fixture("nmap_two_hosts.xml") + hosts = parse_nmap_xml(xml) + assert len(hosts) == 2 + + @pytest.mark.integration + def test_router_host_fields(self): + xml = _fixture("nmap_two_hosts.xml") + router = next(h for h in parse_nmap_xml(xml) if h.ip == "192.168.1.1") + + assert router.mac == "AA:BB:CC:DD:EE:01" + assert router.hostname == "router.local" + assert router.os_guess == "Linux 5.4" + + @pytest.mark.integration + def test_router_open_ports_only(self): + """Closed port 443 must not appear in results.""" + xml = _fixture("nmap_two_hosts.xml") + router = next(h for h in parse_nmap_xml(xml) if h.ip == "192.168.1.1") + + port_nums = [p.port_number for p in router.ports] + assert 22 in port_nums + assert 80 in port_nums + assert 443 not in port_nums # closed — must be excluded + + @pytest.mark.integration + def test_router_ssh_port_details(self): + xml = _fixture("nmap_two_hosts.xml") + router = next(h for h in parse_nmap_xml(xml) if h.ip == "192.168.1.1") + ssh = next(p for p in router.ports if p.port_number == 22) + + assert ssh.protocol == "tcp" + assert ssh.state == "open" + assert ssh.service_name == "ssh" + assert "OpenSSH" in ssh.version_banner + assert "Ubuntu" in ssh.version_banner + + @pytest.mark.integration + def test_workstation_no_mac_no_os(self): + xml = _fixture("nmap_two_hosts.xml") + ws = next(h for h in parse_nmap_xml(xml) if h.ip == "192.168.1.10") + + assert ws.mac == "" + assert ws.os_guess == "" + assert ws.hostname == "" + + @pytest.mark.integration + def test_workstation_smb_port(self): + xml = _fixture("nmap_two_hosts.xml") + ws = next(h for h in parse_nmap_xml(xml) if h.ip == "192.168.1.10") + + assert len(ws.ports) == 1 + assert ws.ports[0].port_number == 445 + assert "WORKGROUP" in ws.ports[0].version_banner + + @pytest.mark.integration + def test_minimal_host_fixture(self): + """Host with no open ports, no hostname, no OS — still included (state=up).""" + xml = _fixture("nmap_minimal_host.xml") + hosts = parse_nmap_xml(xml) + + assert len(hosts) == 1 + assert hosts[0].ip == "192.168.1.50" + assert hosts[0].mac == "DE:AD:BE:EF:00:01" + assert hosts[0].ports == [] # only filtered port — not included + + @pytest.mark.integration + def test_host_down_excluded(self): + """Hosts with state=down must be excluded entirely.""" + xml = _fixture("nmap_host_down.xml") + hosts = parse_nmap_xml(xml) + assert hosts == [] + + @pytest.mark.integration + def test_no_ports_element_returns_empty_ports(self): + """Host with no element at all must parse cleanly with empty ports list.""" + xml = _fixture("nmap_no_ports_element.xml") + hosts = parse_nmap_xml(xml) + assert len(hosts) == 1 + assert hosts[0].ip == "192.168.1.77" + assert hosts[0].ports == [] + + @pytest.mark.integration + def test_os_highest_accuracy_wins(self): + """When multiple osmatch entries exist, the highest accuracy must be chosen.""" + xml = _fixture("nmap_two_hosts.xml") + router = next(h for h in parse_nmap_xml(xml) if h.ip == "192.168.1.1") + # Fixture has accuracy 95 (Linux 5.4) and 80 (Linux 4.15) + assert router.os_guess == "Linux 5.4" + + +# ═══════════════════════════════════════════════════════════════════════════════ +# nmap_scan — unit tests (mocked subprocess) +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestRunNmapScan: + """run_nmap_scan() mocks subprocess.run; XML writing verified via fixture.""" + + @pytest.mark.unit + def test_returns_empty_list_for_empty_hosts(self): + assert run_nmap_scan(hosts=[]) == [] + + @pytest.mark.unit + def test_parses_xml_written_by_nmap(self, tmp_path): + """ + Simulate nmap writing XML to the temp file path captured by the mock. + We intercept the call, copy a fixture to the temp file, then return rc=0. + """ + fixture_xml = _fixture("nmap_two_hosts.xml") + + def fake_run(cmd, **kwargs): + # Find the -oX argument and write fixture content to that path + idx = cmd.index("-oX") + xml_path = cmd[idx + 1] + Path(xml_path).write_text(fixture_xml) + return MagicMock(returncode=0, stdout="", stderr="") + + with patch("app.scanner.nmap_scan.subprocess.run", side_effect=fake_run): + hosts = run_nmap_scan(hosts=["192.168.1.1", "192.168.1.10"], interface="eth0") + + assert len(hosts) == 2 + assert hosts[0].ip == "192.168.1.1" + + @pytest.mark.unit + def test_raises_on_nmap_failure(self): + mock_result = MagicMock(returncode=1, stdout="", stderr="nmap: no interfaces") + with patch("app.scanner.nmap_scan.subprocess.run", return_value=mock_result): + with pytest.raises(RuntimeError, match="nmap failed"): + run_nmap_scan(hosts=["192.168.1.1"]) + + @pytest.mark.unit + def test_temp_file_cleaned_up_on_success(self): + """Temporary XML file must be deleted even on success.""" + fixture_xml = _fixture("nmap_two_hosts.xml") + captured_paths: list[str] = [] + + def fake_run(cmd, **kwargs): + idx = cmd.index("-oX") + xml_path = cmd[idx + 1] + captured_paths.append(xml_path) + Path(xml_path).write_text(fixture_xml) + return MagicMock(returncode=0, stdout="", stderr="") + + with patch("app.scanner.nmap_scan.subprocess.run", side_effect=fake_run): + run_nmap_scan(hosts=["192.168.1.1"]) + + assert captured_paths, "No temp path was captured" + assert not Path(captured_paths[0]).exists(), "Temp XML file was not cleaned up" + + @pytest.mark.unit + def test_temp_file_cleaned_up_on_failure(self): + """Temporary XML file must be deleted even when nmap exits non-zero.""" + captured_paths: list[str] = [] + + def fake_run(cmd, **kwargs): + idx = cmd.index("-oX") + captured_paths.append(cmd[idx + 1]) + return MagicMock(returncode=1, stdout="", stderr="error") + + with patch("app.scanner.nmap_scan.subprocess.run", side_effect=fake_run): + with pytest.raises(RuntimeError): + run_nmap_scan(hosts=["192.168.1.1"]) + + assert captured_paths + assert not Path(captured_paths[0]).exists(), "Temp XML file leaked on failure" + + @pytest.mark.unit + def test_uses_env_var_for_interface(self, monkeypatch): + monkeypatch.setenv("NETWORK_INTERFACE", "wlan0") + fixture_xml = _fixture("nmap_two_hosts.xml") + + def fake_run(cmd, **kwargs): + idx = cmd.index("-oX") + Path(cmd[idx + 1]).write_text(fixture_xml) + return MagicMock(returncode=0, stdout="", stderr="") + + with patch("app.scanner.nmap_scan.subprocess.run", side_effect=fake_run) as mock_run: + run_nmap_scan(hosts=["192.168.1.1"]) + + cmd = mock_run.call_args[0][0] + assert "wlan0" in cmd + + +# ═══════════════════════════════════════════════════════════════════════════════ +# orchestrate_scan — unit tests +# ═══════════════════════════════════════════════════════════════════════════════ + + +class TestOrchestrateScan: + """orchestrate_scan() mocks both run_arp_scan and run_nmap_scan.""" + + @pytest.mark.unit + def test_returns_empty_scan_result_when_no_arp_hosts(self): + with patch("app.scanner.run_arp_scan", return_value=[]): + result = orchestrate_scan() + assert isinstance(result, ScanResult) + assert result.hosts == [] + assert result.arp_only == [] + + @pytest.mark.unit + def test_merges_mac_from_arp_into_nmap(self): + arp_hosts = [ArpHost(ip="192.168.1.1", mac="aa:bb:cc:dd:ee:ff", vendor="Cisco")] + nmap_hosts = [NmapHost(ip="192.168.1.1", mac="")] # nmap didn't capture MAC + + with ( + patch("app.scanner.run_arp_scan", return_value=arp_hosts), + patch("app.scanner.run_nmap_scan", return_value=nmap_hosts), + ): + result = orchestrate_scan() + + assert result.hosts[0].mac == "aa:bb:cc:dd:ee:ff" + assert result.arp_only == [] + + @pytest.mark.unit + def test_keeps_existing_mac_from_nmap(self): + """If nmap already has a MAC, the arp value should not overwrite it.""" + arp_hosts = [ArpHost(ip="192.168.1.1", mac="11:22:33:44:55:66", vendor="")] + nmap_hosts = [NmapHost(ip="192.168.1.1", mac="aa:bb:cc:dd:ee:ff")] + + with ( + patch("app.scanner.run_arp_scan", return_value=arp_hosts), + patch("app.scanner.run_nmap_scan", return_value=nmap_hosts), + ): + result = orchestrate_scan() + + assert result.hosts[0].mac == "aa:bb:cc:dd:ee:ff" + + @pytest.mark.unit + def test_arp_only_hosts_excluded_from_nmap_results(self): + arp_hosts = [ + ArpHost(ip="192.168.1.1", mac="aa:00:00:00:00:01", vendor=""), + ArpHost(ip="192.168.1.2", mac="aa:00:00:00:00:02", vendor=""), + ] + # nmap only returned the first host + nmap_hosts = [NmapHost(ip="192.168.1.1", mac="aa:00:00:00:00:01")] + + with ( + patch("app.scanner.run_arp_scan", return_value=arp_hosts), + patch("app.scanner.run_nmap_scan", return_value=nmap_hosts), + ): + result = orchestrate_scan() + + assert len(result.hosts) == 1 + assert len(result.arp_only) == 1 + assert result.arp_only[0].ip == "192.168.1.2" + + @pytest.mark.unit + def test_passes_interface_and_subnet_through(self): + with ( + patch("app.scanner.run_arp_scan", return_value=[]) as mock_arp, + ): + orchestrate_scan(interface="eth1", subnet="10.10.0.0/16") + + mock_arp.assert_called_once_with(interface="eth1", subnet="10.10.0.0/16") + + @pytest.mark.unit + def test_uses_env_vars_for_interface_and_subnet(self, monkeypatch): + monkeypatch.setenv("NETWORK_INTERFACE", "bond0") + monkeypatch.setenv("SCAN_SUBNET", "172.16.0.0/12") + + with patch("app.scanner.run_arp_scan", return_value=[]) as mock_arp: + orchestrate_scan() + + mock_arp.assert_called_once_with(interface="bond0", subnet="172.16.0.0/12") From b451f5d7c6efaa30b9dccfc9b6d2bea5c3523116 Mon Sep 17 00:00:00 2001 From: wind Date: Fri, 27 Feb 2026 19:55:53 +0100 Subject: [PATCH 2/4] fix: pin ruff to 0.15.x and ignore N817 to fix CI lint failures CI was running ruff==0.4.x which had reversed PT023 opinion (requires parens) and flagged S603 noqa comments differently. Pinning to 0.15.x matches local toolchain. Add N817 to ignore list since 'ET' is the universal stdlib alias for xml.etree.ElementTree. --- backend/pyproject.toml | 2 +- backend/requirements.txt | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/pyproject.toml b/backend/pyproject.toml index 0d9ffd3..2358128 100644 --- a/backend/pyproject.toml +++ b/backend/pyproject.toml @@ -17,7 +17,7 @@ src = ["app"] [tool.ruff.lint] select = ["E", "F", "W", "I", "N", "UP", "S", "B", "A", "C4", "PT"] -ignore = ["S101"] # allow assert in tests +ignore = ["S101", "N817"] # allow assert in tests; ET alias is universal stdlib convention [tool.ruff.lint.per-file-ignores] "tests/*" = ["S", "ANN"] diff --git a/backend/requirements.txt b/backend/requirements.txt index d0462f2..cd04843 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -18,7 +18,7 @@ pytest-asyncio==0.23.* httpx==0.27.* # async test client for FastAPI # Linting / formatting -ruff==0.4.* +ruff==0.15.* # Security audit pip-audit==2.* From d88783527c4bab98192c082d839a4204f809b339 Mon Sep 17 00:00:00 2001 From: wind Date: Fri, 27 Feb 2026 19:56:59 +0100 Subject: [PATCH 3/4] fix: add justification comment to noqa in app/__init__.py --- backend/app/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/app/__init__.py b/backend/app/__init__.py index 3023a03..a76332d 100644 --- a/backend/app/__init__.py +++ b/backend/app/__init__.py @@ -1,5 +1,5 @@ """Package init — exposes the FastAPI app for uvicorn.""" -from app.main import app # noqa: F401 +from app.main import app # noqa: F401 — re-exported for uvicorn app:app entrypoint __all__ = ["app"] From 39460e7b2727459b1ff1f23abf5ce1c828a70cb7 Mon Sep 17 00:00:00 2001 From: wind Date: Fri, 27 Feb 2026 19:58:50 +0100 Subject: [PATCH 4/4] fix: bump fastapi to 0.120.x and floor starlette>=0.47.2 to resolve CVEs CVE-2024-47874 and CVE-2025-54121 in starlette 0.37.x; fix is starlette>=0.47.2 which requires fastapi>=0.120 for dependency compatibility. --- backend/requirements.txt | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/backend/requirements.txt b/backend/requirements.txt index cd04843..f654a50 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -2,7 +2,8 @@ # Pin major versions here; exact pins are locked in requirements.lock (generated by pip-compile) # Web framework -fastapi==0.111.* +fastapi==0.120.* +starlette>=0.47.2 # pin floor to resolve CVE-2024-47874 and CVE-2025-54121 uvicorn[standard]==0.29.* # Database