admin / Synapse-NetscanXi

"""
nmap execution: scan profiles, live progress parsing, cancellation.

Performance model
-----------------
For network ranges we run a *two-phase* scan:

  1. Discovery: a fast ping/ARP sweep (`-sn`) to find which hosts are actually
     up. This is cheap and avoids wasting the expensive scan on dead IPs.
  2. Targeted: the heavy scan (service/OS/vuln) runs ONLY against the live
     hosts from phase 1.

On a typical /24 with a handful of live hosts this is dramatically faster than
scanning all 254 addresses with `-O`/`-sV`/NSE, because nmap no longer spends
time probing (and waiting on retransmits for) hosts that will never answer.

Single-IP targets skip discovery and scan directly.

Each profile also carries timing tuning (`--min-rate`, `--max-retries`,
`--host-timeout`, `-n`) that meaningfully reduces wall-clock time without
materially hurting result quality on a LAN.
"""

import ipaddress
import os
import re
import signal
import subprocess
import tempfile

PROGRESS_RE = re.compile(r"About\s+([\d.]+)%\s+done")
# v8: capture the host nmap is currently working on so the UI can display it.
#   "Scanning 192.168.1.5 [1000 ports]"  (single)
#   "Scanning 4 hosts [...]"             (batch - ignored, no single IP)
#   "Nmap scan report for host (192.168.1.5)" / "... for 192.168.1.5"
_IP_RE = re.compile(r"(\d{1,3}(?:\.\d{1,3}){3})")
_SCANNING_RE = re.compile(r"Scanning\s+(.+?)\s+\[")
_REPORT_RE = re.compile(r"Nmap scan report for\s+(.+)")

# Common timing flags applied to the heavy (phase 2) scan.
#   -n               : skip reverse-DNS on every host (big saver on large ranges)
#   --max-retries 2  : fewer probe retransmits to filtered ports
#   --min-rate 500   : keep the packet rate up so we are not idle-waiting
#   --defeat-rst-ratelimit : don't let RST rate-limiting stall closed-port detection
PERF_FLAGS = ["-n", "--max-retries", "2", "--min-rate", "500",
              "--defeat-rst-ratelimit"]

# Per-host timeout so one slow/filtered host can't drag out the whole scan.
HOST_TIMEOUT = "10m"

# Scan profiles: tradeoff between speed and depth.
# `args` are the *detection* flags; timing/perf flags are added by build_command.
PROFILES = {
    "quick": {
        "label": "Quick (top 100 ports, fast)",
        "args": ["-F", "-sV", "--version-light", "-T4", "--osscan-limit"],
        "vuln": False,
        "discovery": True,
    },
    "standard": {
        "label": "Standard (top 1000 ports, service + OS)",
        "args": ["-sS", "-sV", "-O", "-T4", "--osscan-limit", "--max-os-tries", "1"],
        "vuln": False,
        "discovery": True,
    },
    "deep": {
        "label": "Deep (service + OS + vulnerability scripts)",
        # v9.1: add the version-aware `vulners` script alongside the generic
        # `vuln` category so CVEs are matched against the exact detected
        # product+version (and captured patch token) rather than only the
        # broad distro release — far fewer false positives/negatives.
        "args": ["-sS", "-sV", "--version-all", "-sC", "-O", "-T4",
                 "--osscan-guess", "--script", "vuln,vulners"],
        "vuln": True,
        "discovery": True,
    },
}

# v9.1: NSE scripts that perform patch/version-aware CVE matching. `vulners`
# correlates the detected CPE + version against its CVE database; the http-* and
# smb-vuln helpers add patch-state checks for common services.
PATCH_AWARE_SCRIPTS = ["vulners"]
DEFAULT_PROFILE = "standard"

# ---------------------------------------------------------------------------
# Scan options (v3): user-selectable toggles that shape the nmap command.
# Defaults keep parity with the chosen profile.
# ---------------------------------------------------------------------------
DEFAULT_OPTIONS = {
    "ports": True,        # scan ports at all
    "services": True,     # -sV service/version detection
    "os": True,           # -O OS detection
    "docker": True,       # probe for Docker hosts (adds docker ports + scripts)
    "docker_deep": True,  # v10r1: authenticated API inventory + image CVE + CIS audit
    "vulns": False,       # --script vuln (slow); profile may force-enable
    "patch_audit": True,  # v9.1: version+patch-aware CVE matching (vulners NSE)
    "creds": False,       # credential-EXPOSURE detection (defensive; see below)
    "ad_audit": False,    # v7: Active Directory / Domain Controller auditing
    "ssl_certs": False,   # v7: deep SSL/TLS certificate expiry tracking
}

# v7: Active Directory / Domain Controller auditing - DEFENSIVE enumeration.
# Identifies DCs and reports directory/auth posture (SMB signing, LDAP info,
# Kerberos, supported SMB protocol versions). No password attacks / no *-brute.
AD_PORTS = ["88", "389", "636", "445", "3268", "3269", "464", "53"]
AD_SCRIPTS = [
    "smb-os-discovery",       # host/domain/forest info
    "smb-security-mode",      # SMB signing posture
    "smb2-security-mode",     # SMBv2/3 signing posture
    "smb-protocols",          # which SMB dialects are offered (SMBv1!)
    "ldap-rootdse",           # directory naming contexts (DC fingerprint)
    "krb5-enum-users",        # NOTE: enumeration only, no password guessing
]

# v7: deep SSL/TLS certificate + cipher inspection for expiry tracking.
SSL_SCRIPTS = ["ssl-cert", "ssl-enum-ciphers"]

# Credential-exposure detection (v6) - DEFENSIVE ONLY.
# These NSE scripts identify weak/cleartext authentication and missing transport
# encryption so they can be remediated. They do NOT brute-force, capture, or
# guess credentials. (We deliberately exclude *-brute and password-grabbing
# scripts.) Pairs with app/compliance.py for the risk + framework mapping.
CREDS_SCRIPTS = [
    "ssl-enum-ciphers",   # weak/no TLS on a port
    "http-auth",          # HTTP auth scheme in use (e.g. Basic over cleartext)
    "ftp-anon",           # anonymous/odd FTP auth exposure
    "ssh-auth-methods",   # which SSH auth methods are offered (e.g. password)
    "rdp-ntlm-info",      # RDP exposure / NLA posture
    "banner",             # service banners to spot cleartext login services
]

# Ports that commonly indicate a Docker host / container runtime:
#   2375 Docker API (plain), 2376 Docker API (TLS), 2377 Swarm mgmt,
#   7946 Swarm gossip, 4243 legacy Docker API, 5000 common registry,
#   2379/2380 etcd (often k8s/docker swarm), 6443 kube-api,
#   10250 kubelet, 8080 dashboards/portainer-ish.
DOCKER_PORTS = ["2375", "2376", "2377", "4243", "7946", "2379", "2380",
                "5000", "6443", "10250"]

# Fast discovery sweep: ping scan, no port scan, no DNS. Quick ARP/ICMP on a LAN.
DISCOVERY_ARGS = ["-sn", "-n", "-T4", "--min-rate", "1000", "--max-retries", "1"]

PHASE_WEIGHTS = [
    ("Ping Scan", 0, 8),
    ("Parallel DNS resolution", 8, 10),
    ("SYN Stealth Scan", 10, 40),
    ("Connect Scan", 10, 40),
    ("Service scan", 40, 62),
    ("OS detection", 62, 72),
    ("NSE", 72, 99),
    ("Script Scan", 72, 99),
]


def _phase_bounds(label):
    for name, lo, hi in PHASE_WEIGHTS:
        if label.startswith(name):
            return name, lo, hi
    return None, None, None


def _is_range(target):
    """True if target covers more than one address (CIDR or dash range)."""
    t = (target or "").strip()
    if "/" in t:
        try:
            return ipaddress.ip_network(t, strict=False).num_addresses > 1
        except ValueError:
            return False
    # nmap dash ranges / comma lists / wildcards => treat as multi-host
    return any(c in t for c in ("-", ",", "*"))


def build_discovery_command(target):
    return ["nmap", *DISCOVERY_ARGS, "-oG", "-", target]


def normalize_options(profile, options):
    """Merge user options with profile defaults into a complete option dict."""
    prof = PROFILES.get(profile, PROFILES[DEFAULT_PROFILE])
    opts = dict(DEFAULT_OPTIONS)
    # Profile sets the vuln baseline; user can still override below.
    opts["vulns"] = bool(prof.get("vuln"))
    if options:
        for k in DEFAULT_OPTIONS:
            if k in options:
                opts[k] = bool(options[k])
    return opts


def build_command(target_args, profile, xml_path, options=None):
    """Build an nmap command from a profile + selectable scan options.

    `target_args` is a list of trailing nmap target tokens, e.g.
    ["192.168.1.0/24"] or ["-iL", "/tmp/targets.txt"].
    """
    opts = normalize_options(profile, options)
    prof = PROFILES.get(profile, PROFILES[DEFAULT_PROFILE])

    args = []
    # Base scan-type / port flags from the profile, filtered by toggles.
    for a in prof["args"]:
        if a == "-sV" and not opts["services"]:
            continue
        if a in ("-O",) and not opts["os"]:
            continue
        if a in ("--osscan-limit", "--osscan-guess", "--max-os-tries") and not opts["os"]:
            continue
        if a in ("--script", "vuln") and not opts["vulns"]:
            continue
        # version-detection sub-flags only make sense with -sV
        if a in ("--version-light", "--version-all") and not opts["services"]:
            continue
        args.append(a)

    # If the user disabled port scanning entirely, do a host/OS-only scan.
    if not opts["ports"]:
        args = [a for a in args if a not in ("-F", "-sS")]
        if "-sn" not in args:
            args.insert(0, "-sn")

    # Docker detection: ensure the docker-indicator ports are included and add
    # the docker/http NSE scripts that fingerprint the API.
    if opts["docker"] and opts["ports"]:
        args += ["-p", _merge_docker_ports(args)]
        # docker-version probes the remote Docker API; http-* help on dashboards.
        if opts["services"]:
            args += ["--script", "docker-version"]

    # Re-add vuln scripts if user force-enabled on a non-deep profile.
    if opts["vulns"] and "--script" not in args:
        args += ["--script", "vuln"]
    elif opts["vulns"] and "vuln" not in args:
        # a --script already present (e.g. docker-version): append vuln
        try:
            i = args.index("--script")
            args[i + 1] = args[i + 1] + ",vuln"
        except (ValueError, IndexError):
            args += ["--script", "vuln"]

    def _add_scripts(spec):
        """Merge a comma-joined NSE script spec into the command."""
        if "--script" in args:
            i = args.index("--script")
            args[i + 1] = args[i + 1] + "," + spec
        else:
            args.extend(["--script", spec])

    # v9.1: patch/version-aware CVE matching (vulners). Needs version detection
    # and an open-port scan to be useful; merges into any existing --script.
    if opts.get("patch_audit") and opts["vulns"] and opts["ports"]:
        spec = ",".join(PATCH_AWARE_SCRIPTS)
        if "--script" in args:
            i = args.index("--script")
            if "vulners" not in args[i + 1]:
                args[i + 1] = args[i + 1] + "," + spec
        else:
            args.extend(["--script", spec])
        if "-sV" not in args:
            args.insert(0, "-sV")

    # Credential-exposure detection (v6, defensive). Needs an open port scan and
    # service detection to be meaningful. Merges into any existing --script.
    if opts.get("creds") and opts["ports"]:
        _add_scripts(",".join(CREDS_SCRIPTS))
        if "-sV" not in args:
            args.insert(0, "-sV")  # banners/auth scripts need version probes

    # v7: Active Directory / Domain Controller auditing (defensive enumeration).
    if opts.get("ad_audit") and opts["ports"]:
        _add_scripts(",".join(AD_SCRIPTS))
        merged = _merge_extra_ports(args, AD_PORTS)
        if "-p" in args:
            args[args.index("-p") + 1] = merged   # replace, don't duplicate
        else:
            args += ["-p", merged]
        if "-sV" not in args:
            args.insert(0, "-sV")

    # v7: deep SSL/TLS certificate expiry tracking.
    if opts.get("ssl_certs") and opts["ports"]:
        _add_scripts(",".join(SSL_SCRIPTS))
        if "-sV" not in args:
            args.insert(0, "-sV")

    cmd = ["nmap", *args, *PERF_FLAGS,
           "--host-timeout", HOST_TIMEOUT,
           "--stats-every", "1s", "-oX", xml_path, *target_args]
    return cmd


def _merge_extra_ports(args, extra_ports):
    """Return a -p spec covering the profile's base range plus extra ports.
    If a -p already exists in args, extend it; otherwise build from the base."""
    base = "1-100" if "-F" in args else "1-1000"
    existing = []
    if "-p" in args:
        try:
            existing = args[args.index("-p") + 1].split(",")
        except (ValueError, IndexError):
            existing = []
    ports = (existing or base.split(",")) + list(extra_ports)
    seen, out = set(), []
    for p in ports:
        if p and p not in seen:
            seen.add(p)
            out.append(p)
    return ",".join(out)


def _merge_docker_ports(args):
    """Return a -p port spec that keeps the default top ports plus docker ports.
    nmap has no 'top-ports + extra' syntax in one -p, so we expand explicitly:
    use the profile's implied range plus the docker ports."""
    # If -F (fast/top-100) is in play we approximate with top-100 + docker;
    # otherwise default top-1000. We express this as '1-1000' plus docker ports
    # for determinism, de-duplicated.
    base = "1-1000"
    if "-F" in args:
        base = "1-100"
    ports = base.split(",") + DOCKER_PORTS
    seen, out = set(), []
    for p in ports:
        if p not in seen:
            seen.add(p)
            out.append(p)
    return ",".join(out)


class ScanProcess:
    """Wraps the currently-running nmap process so the caller can cancel it."""

    def __init__(self):
        self.proc = None
        self.cancelled = False

    def set_proc(self, proc):
        self.proc = proc

    def cancel(self):
        self.cancelled = True
        self._kill(self.proc)

    @staticmethod
    def _kill(proc):
        if proc and proc.poll() is None:
            try:
                if os.name == "posix":
                    os.killpg(os.getpgid(proc.pid), signal.SIGTERM)
                else:
                    proc.terminate()
            except (ProcessLookupError, OSError):
                pass


def _popen(cmd):
    popen_kw = dict(stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                    text=True, bufsize=1)
    if os.name == "posix":
        popen_kw["preexec_fn"] = os.setsid  # own process group for clean kill
    return subprocess.Popen(cmd, **popen_kw)


def discover_hosts(target, progress_cb=None, handle=None):
    """
    Fast ping sweep. Returns a list of live host IPs (strings).
    Best-effort: on any failure returns [] so the caller falls back to scanning
    the whole target directly.
    """
    cmd = build_discovery_command(target)
    proc = _popen(cmd)
    if handle is not None:
        handle.set_proc(proc)

    live = []
    host_re = re.compile(r"Host:\s+(\S+)\s+.*Status:\s+Up", re.IGNORECASE)
    if progress_cb:
        progress_cb(1.0, "Host discovery")
    try:
        for line in proc.stdout:
            if handle is not None and handle.cancelled:
                break
            m = host_re.search(line)
            if m:
                live.append(m.group(1))
                if progress_cb:
                    progress_cb(min(8.0, 1.0 + len(live) * 0.2),
                                f"Host discovery ({len(live)} up)")
    finally:
        try:
            proc.wait(timeout=300)
        except subprocess.TimeoutExpired:
            proc.kill()
    return live


def _run_scan_pass(target_args, profile, progress_cb, handle, base_lo=0.0,
                   options=None):
    """Run one heavy nmap pass. `target_args` is a list of trailing nmap
    target tokens (a CIDR/IP, or ["-iL", path]). Parses live progress."""
    fd, xml_path = tempfile.mkstemp(suffix=".xml", prefix="netscan_")
    os.close(fd)
    cmd = build_command(target_args, profile, xml_path, options=options)
    proc = _popen(cmd)
    if handle is not None:
        handle.set_proc(proc)

    current_phase = "Starting scan"
    last_pct = base_lo
    current_ip = None

    def _emit(pct, phase):
        """Forward progress, including the current IP when the callback accepts it."""
        if not progress_cb:
            return
        try:
            progress_cb(pct, phase, current_ip)
        except TypeError:
            progress_cb(pct, phase)

    try:
        for line in proc.stdout:
            if handle is not None and handle.cancelled:
                break
            line = line.strip()
            if not line:
                continue

            # v8: track which host nmap is actively scanning / reporting on.
            m_scan = _SCANNING_RE.search(line)
            if m_scan:
                ip_m = _IP_RE.search(m_scan.group(1))
                if ip_m:
                    current_ip = ip_m.group(1)
                    _emit(last_pct, current_phase)
            else:
                m_rep = _REPORT_RE.search(line)
                if m_rep:
                    ip_m = _IP_RE.search(m_rep.group(1))
                    if ip_m:
                        current_ip = ip_m.group(1)
                        _emit(last_pct, current_phase)

            if line.startswith("Initiating "):
                current_phase = line[len("Initiating "):].split(" at ")[0]
                _, lo, _ = _phase_bounds(current_phase)
                if lo is not None:
                    scaled = base_lo + lo * (1 - base_lo / 100.0)
                    _emit(max(scaled, last_pct), current_phase)
                continue

            if "NSE: Script scanning" in line or line.startswith("NSE:"):
                current_phase = "NSE"
                _emit(max(base_lo + 72.0 * (1 - base_lo / 100.0), last_pct),
                      "Vulnerability & script scan")
                continue

            m = PROGRESS_RE.search(line)
            if m:
                pct = float(m.group(1))
                name, lo, hi = _phase_bounds(current_phase)
                if lo is None:
                    lo, hi = last_pct, min(99.0, last_pct + 10)
                phase_overall = lo + (hi - lo) * (pct / 100.0)
                overall = base_lo + phase_overall * (1 - base_lo / 100.0)
                if overall >= last_pct:
                    last_pct = overall
                    _emit(overall, current_phase)
    finally:
        try:
            proc.wait(timeout=1800)
        except subprocess.TimeoutExpired:
            proc.kill()

    if handle is not None and handle.cancelled:
        _safe_unlink(xml_path)
        return None, proc.returncode

    xml_text = ""
    if os.path.exists(xml_path):
        with open(xml_path, "r", encoding="utf-8", errors="replace") as fh:
            xml_text = fh.read()
    _safe_unlink(xml_path)
    return xml_text, proc.returncode


def run_nmap(target, profile, progress_cb=None, handle=None, options=None):
    """
    Two-phase scan for ranges; direct scan for single hosts.
    `options` is the v3 selectable-options dict (ports/services/os/docker/vulns).
    Returns nmap XML text, or "" if cancelled.
    """
    prof = PROFILES.get(profile, PROFILES[DEFAULT_PROFILE])

    if prof.get("discovery") and _is_range(target):
        live = discover_hosts(target, progress_cb=progress_cb, handle=handle)
        if handle is not None and handle.cancelled:
            return ""
        if live:
            # Scan only the live hosts, passed via -iL (a temp file) to avoid
            # command-line length limits on large host lists.
            list_path = _write_target_list(live)
            if progress_cb:
                progress_cb(8.0, f"Scanning {len(live)} live host(s)")
            try:
                xml, rc = _run_scan_pass(["-iL", list_path], profile,
                                         progress_cb, handle, base_lo=8.0,
                                         options=options)
            finally:
                _safe_unlink(list_path)
            return _finalize(xml, rc, handle)
        # No live hosts found (or discovery failed) -> fall back to full scan.

    xml, rc = _run_scan_pass([target], profile, progress_cb, handle, base_lo=0.0,
                             options=options)
    return _finalize(xml, rc, handle)


def _write_target_list(ips):
    fd, path = tempfile.mkstemp(suffix=".txt", prefix="netscan_targets_")
    with os.fdopen(fd, "w", encoding="utf-8") as fh:
        fh.write("\n".join(ips) + "\n")
    return path


def _finalize(xml, rc, handle):
    if handle is not None and handle.cancelled:
        return ""
    if rc not in (0, None) and not (xml or "").strip():
        raise RuntimeError(f"nmap failed (exit {rc})")
    return xml or ""


def _safe_unlink(path):
    try:
        os.unlink(path)
    except OSError:
        pass