[Phase 2] arp-scan + nmap integration and XML parsing

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"]

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)

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<ip>\d{1,3}(?:\.\d{1,3}){3})\t(?P<mac>[0-9a-fA-F:]{17})(?:\t(?P<vendor>.*))?$"
+)
+
+
+@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

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

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"]

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.*

backend/tests/fixtures/nmap_host_down.xml

@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE nmaprun>
+<!-- Edge-case fixture: host with status="down" — should be excluded by parser -->
+<nmaprun scanner="nmap" args="nmap -sV -O --top-ports 1000 -e eth0 -oX /tmp/out.xml 192.168.1.99"
+         start="1700000200" version="7.94" xmloutputversion="1.04">
+
+  <host starttime="1700000201" endtime="1700000202">
+    <status state="down" reason="no-response" reason_ttl="0"/>
+    <address addr="192.168.1.99" addrtype="ipv4"/>
+    <ports/>
+  </host>
+
+</nmaprun>