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"] 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/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..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 @@ -18,7 +19,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.* 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")