admin / Synapse-NetscanXi

"""
nmap XML -> rich host dicts, plus risk scoring and device-type guessing.
CVE enrichment (NVD + CISA KEV) lives in enrich.py and is applied afterwards.
"""

import ipaddress
import re
import xml.etree.ElementTree as ET

from . import compliance, auditors

CVE_RE = re.compile(r"(CVE-\d{4}-\d{4,7})", re.IGNORECASE)

# v9.1: patch / build identifiers we try to lift out of version strings and
# service banners so vulnerability assessment is version+patch aware rather than
# only distro-version aware. Examples captured:
#   "OpenSSH 8.2p1 Ubuntu-4ubuntu0.5"  -> patch "4ubuntu0.5"
#   "Apache httpd 2.4.41 ((Ubuntu))"   -> distro tag "Ubuntu"
#   "... build 1234" / "... patch 7"   -> build/patch number
PATCH_PATTERNS = [
    re.compile(r"\b(?:patch(?:\s*level)?|sp|service\s*pack)\s*[:\-]?\s*(\d+)\b", re.I),
    re.compile(r"\b(?:build)\s*[:\-]?\s*([0-9][0-9A-Za-z.\-_]*)\b", re.I),
    re.compile(r"(\d+(?:ubuntu|debian|deb|el|rhel|fc|\.fc|\+deb)[0-9A-Za-z.\-_+~]*)", re.I),
    re.compile(r"-(\d+\.[0-9A-Za-z.\-_+~]+)\b"),  # rpm/deb release suffix
]


def extract_patch_info(product, version, extrainfo):
    """v9.1: best-effort extraction of a patch / build / package-release token
    from nmap's product/version/extrainfo. Returns '' when nothing is found.
    This lets the risk + enrichment layers reason about *patched* releases
    instead of just the base upstream version."""
    blob = " ".join(p for p in (version or "", extrainfo or "") if p).strip()
    if not blob:
        return ""
    for pat in PATCH_PATTERNS:
        m = pat.search(blob)
        if m:
            return m.group(1)
    return ""

# Docker / container-runtime indicators.
# Port -> what it suggests. Used to flag a host as a likely Docker host.
DOCKER_PORT_HINTS = {
    "2375": "Docker API (unencrypted)",
    "2376": "Docker API (TLS)",
    "2377": "Docker Swarm management",
    "4243": "Docker API (legacy)",
    "7946": "Docker Swarm gossip",
    "2379": "etcd (Swarm/k8s)",
    "2380": "etcd peer (Swarm/k8s)",
    "6443": "Kubernetes API",
    "10250": "Kubelet",
}
# nmap service/product strings that betray Docker.
DOCKER_SERVICE_HINTS = ("docker", "moby", "containerd", "kubelet", "kubernetes")

# Rough OUI/keyword + port heuristics for device-type guessing.
VENDOR_HINTS = [
    ("router", ["mikrotik", "ubiquiti", "tp-link", "netgear", "asus", "draytek", "cisco", "zyxel"]),
    ("camera", ["hikvision", "dahua", "axis", "reolink", "amcrest", "wyze"]),
    ("printer", ["hewlett", "hp", "canon", "epson", "brother", "lexmark"]),
    ("nas", ["synology", "qnap", "western digital", "netapp"]),
    ("iot", ["espressif", "tuya", "sonoff", "shelly", "nest", "ring", "amazon technologies"]),
    ("phone", ["apple", "samsung", "google", "huawei", "xiaomi", "oneplus"]),
    ("tv/media", ["roku", "lg electronics", "sony", "vizio", "sonos"]),
    ("raspberry pi", ["raspberry pi"]),
]
PORT_HINTS = [
    ("printer", {515, 631, 9100}),
    ("camera", {554, 8000, 8554}),
    ("nas", {139, 445, 548, 873, 5000, 5001}),
    ("router", {53, 67, 1900, 8291}),
    ("web server", {80, 443, 8080, 8443}),
    ("database", {3306, 5432, 1433, 27017, 6379}),
]


def guess_device_type(host):
    vendor = (host.get("vendor") or "").lower()
    os_name = (host.get("os") or "").lower()
    for dtype, kws in VENDOR_HINTS:
        if any(k in vendor for k in kws):
            return dtype
    open_ports = set()
    for pd in host.get("port_details", []):
        try:
            open_ports.add(int(pd["port"]))
        except (ValueError, KeyError, TypeError):
            pass
    for dtype, ports in PORT_HINTS:
        if open_ports & ports:
            return dtype
    if "windows" in os_name:
        return "windows host"
    if "linux" in os_name:
        return "linux host"
    return "unknown"


def _detect_docker(port_details):
    """
    Decide whether a host is running Docker / a container runtime, based on
    open docker-indicator ports, service banners, and the docker-version NSE
    script output. Returns a dict:
        {is_docker, confidence, indicators[], api_exposed, version, ports[]}
    """
    indicators = []
    docker_ports = []
    api_exposed = False
    version = ""

    for pd in port_details:
        port = str(pd.get("port", ""))
        svc = (pd.get("service") or "").lower()
        product = (pd.get("product") or "").lower()

        if port in DOCKER_PORT_HINTS:
            docker_ports.append(port)
            indicators.append(f"port {port} — {DOCKER_PORT_HINTS[port]}")
            if port in ("2375", "2376", "4243"):
                api_exposed = True

        blob = f"{svc} {product}"
        if any(k in blob for k in DOCKER_SERVICE_HINTS):
            indicators.append(f"port {port} service looks like Docker ({svc or product})")
            api_exposed = api_exposed or port in ("2375", "2376", "4243")

        # docker-version NSE script output carries the engine version.
        for sc in pd.get("scripts", []):
            sid = (sc.get("id") or "").lower()
            out = sc.get("output", "") or ""
            if "docker" in sid:
                indicators.append(f"NSE {sc.get('id')} on port {port}")
                m = re.search(r"[Vv]ersion[:\s]+([0-9][0-9A-Za-z.\-_]+)", out)
                if m and not version:
                    version = m.group(1)
                api_exposed = True

    is_docker = bool(indicators)
    # Confidence: API banner/script = high; multiple ports = medium; single = low.
    if version or api_exposed:
        confidence = "high"
    elif len(docker_ports) >= 2 or (docker_ports and len(indicators) >= 2):
        confidence = "medium"
    elif docker_ports:
        confidence = "low"
    else:
        confidence = "none"

    return {
        "is_docker": is_docker,
        "confidence": confidence,
        "indicators": indicators,
        "api_exposed": api_exposed,
        "version": version,
        "ports": docker_ports,
    }


def _extract_vulns_from_script(script_el):
    vulns = []
    sid = script_el.get("id", "")
    output = script_el.get("output", "") or ""

    structured = False
    for table in script_el.iter("table"):
        entry = {}
        for elem in table.findall("elem"):
            key = elem.get("key", "")
            if key:
                entry[key] = (elem.text or "").strip()
        vid = entry.get("id") or entry.get("cve") or ""
        if vid and CVE_RE.match(vid):
            structured = True
            vulns.append({
                "id": vid.upper(),
                "cvss": entry.get("cvss") or entry.get("cvss_score") or "",
                "source": sid,
                "exploit": entry.get("is_exploit", "").lower() in ("true", "1", "yes"),
            })

    if not structured:
        seen = set()
        for cve in CVE_RE.findall(output):
            cve = cve.upper()
            if cve not in seen:
                seen.add(cve)
                vulns.append({"id": cve, "cvss": "", "source": sid, "exploit": False})

    if not vulns and "VULNERABLE" in output.upper():
        first = output.strip().splitlines()[0] if output.strip() else sid
        vulns.append({"id": f"{sid}: {first[:80]}", "cvss": "",
                      "source": sid, "exploit": False})
    return vulns


def compute_risk(host):
    """0-100 risk score from vulns, exposure, and dangerous services."""
    score = 0.0
    for v in host.get("vulns", []):
        try:
            cvss = float(v.get("cvss") or 0)
        except (TypeError, ValueError):
            cvss = 0.0
        contrib = cvss * 3 if cvss else 6  # unknown-severity vuln still counts
        if v.get("kev"):
            contrib += 25
        elif v.get("exploit"):
            contrib += 12
        score += contrib

    open_ports = host.get("port_details", [])
    score += min(len(open_ports) * 1.5, 15)

    risky = {21: "ftp", 23: "telnet", 3389: "rdp", 445: "smb",
             5900: "vnc", 1900: "upnp", 161: "snmp"}
    for pd in open_ports:
        try:
            if int(pd["port"]) in risky:
                score += 5
        except (ValueError, KeyError, TypeError):
            pass

    return int(max(0, min(100, round(score))))


def parse_nmap_xml(xml_text):
    hosts = []
    if not xml_text.strip():
        return hosts
    root = ET.fromstring(xml_text)

    for host in root.findall("host"):
        status = host.find("status")
        if status is not None and status.get("state") != "up":
            continue

        ip = mac = vendor = ""
        for addr in host.findall("address"):
            atype = addr.get("addrtype")
            if atype in ("ipv4", "ipv6"):
                ip = addr.get("addr", "")
            elif atype == "mac":
                mac = addr.get("addr", "")
                vendor = addr.get("vendor", "")

        hostname = ""
        hostnames = host.find("hostnames")
        if hostnames is not None:
            hn = hostnames.find("hostname")
            if hn is not None:
                hostname = hn.get("name", "")

        os_name = os_accuracy = ""
        os_el = host.find("os")
        if os_el is not None:
            match = os_el.find("osmatch")
            if match is not None:
                os_name = match.get("name", "")
                os_accuracy = match.get("accuracy", "")

        open_ports, port_details, software, vulns = [], [], [], []
        seen_software = set()

        ports_el = host.find("ports")
        if ports_el is not None:
            for port in ports_el.findall("port"):
                state = port.find("state")
                if state is None or state.get("state") != "open":
                    continue
                portid = port.get("portid", "")
                protocol = port.get("protocol", "")
                service_el = port.find("service")
                svc = product = version = extrainfo = ""
                cpes = []
                if service_el is not None:
                    svc = service_el.get("name", "")
                    product = service_el.get("product", "")
                    version = service_el.get("version", "")
                    extrainfo = service_el.get("extrainfo", "")
                    for cpe in service_el.findall("cpe"):
                        if cpe.text:
                            cpes.append(cpe.text)

                # Enhanced software/application inventory (v6): record every
                # detected application instance with full detail. We key the
                # de-dupe on (product, version, port) so the same app on two
                # ports is still listed, but identical repeats are collapsed.
                soft_label = " ".join(p for p in (product, version) if p).strip()
                ostype = service_el.get("ostype", "") if service_el is not None else ""
                devicetype = service_el.get("devicetype", "") if service_el is not None else ""
                if soft_label or svc:
                    display = soft_label or svc
                    dedupe_key = f"{display.lower()}|{portid}"
                    if dedupe_key not in seen_software:
                        seen_software.add(dedupe_key)
                        # v9.1: capture a patch/build token so assessment is
                        # patch-aware, not just distro-version aware.
                        patch = extract_patch_info(product, version, extrainfo)
                        software.append({
                            "name": display,
                            "product": product,
                            "version": version,
                            "patch": patch,
                            "service": svc,
                            "port": f"{portid}/{protocol}",
                            "extrainfo": extrainfo,
                            "ostype": ostype,
                            "devicetype": devicetype,
                            "cpes": cpes,
                        })

                port_vulns, scripts = [], []
                for script_el in port.findall("script"):
                    sid = script_el.get("id", "")
                    out = (script_el.get("output", "") or "").strip()
                    scripts.append({"id": sid, "output": out})
                    pv = _extract_vulns_from_script(script_el)
                    _patch = extract_patch_info(product, version, extrainfo)
                    for v in pv:
                        v["port"] = f"{portid}/{protocol}"
                        v["service"] = svc
                        v["product"] = product
                        v["version"] = version
                        v["patch"] = _patch
                        # v9.1: a vuln tied to a concrete product+version (and
                        # ideally a patch token) is version-confirmed, not just
                        # a heuristic match against the distro release.
                        v["confirmed"] = bool(product and version)
                    port_vulns.extend(pv)
                vulns.extend(port_vulns)

                svc_label = svc
                extra = " ".join(p for p in (product, version) if p)
                if extra:
                    svc_label = f"{svc} ({extra})"
                open_ports.append(f"{portid}/{protocol} {svc_label}".strip())
                port_details.append({
                    "port": portid, "protocol": protocol, "service": svc,
                    "product": product, "version": version, "extrainfo": extrainfo,
                    "cpes": cpes, "scripts": scripts, "vulns": port_vulns,
                })

        # ---- Docker host detection (v3) ----
        docker = _detect_docker(port_details)

        host_scripts = []
        hostscript_el = host.find("hostscript")
        if hostscript_el is not None:
            for script_el in hostscript_el.findall("script"):
                sid = script_el.get("id", "")
                out = (script_el.get("output", "") or "").strip()
                host_scripts.append({"id": sid, "output": out})
                hv = _extract_vulns_from_script(script_el)
                for v in hv:
                    v["port"] = "host"
                vulns.extend(hv)

        uniq, seen_vid = [], set()
        for v in vulns:
            if v["id"] in seen_vid:
                continue
            seen_vid.add(v["id"])
            uniq.append(v)

        def _cvss_key(v):
            try:
                return float(v.get("cvss") or 0)
            except ValueError:
                return 0.0
        uniq.sort(key=_cvss_key, reverse=True)

        # v9.1: roll up detected patch/build levels across all software so the
        # UI and reports can show how patched a host actually is.
        patch_levels = []
        for sw in software:
            if sw.get("patch"):
                patch_levels.append({"product": sw.get("product") or sw.get("name"),
                                     "version": sw.get("version"),
                                     "patch": sw.get("patch"),
                                     "port": sw.get("port")})

        h = {
            "ip": ip, "hostname": hostname or "-", "mac": mac or "-",
            "vendor": vendor or "-", "os": os_name or "unknown",
            "os_accuracy": os_accuracy, "ports": open_ports,
            "port_details": port_details, "software": software,
            "software_count": len(software),
            "patch_levels": patch_levels,
            "vulns": uniq, "vuln_count": len(uniq), "host_scripts": host_scripts,
            "docker": docker,
        }
        h["device_type"] = guess_device_type(h)
        h["risk"] = compute_risk(h)
        # v7: TLS cert expiry tracking + AD/DC audit (feed compliance).
        h["certificates"] = auditors.certificate_findings(h)
        ad = auditors.ad_audit(h)
        h["ad"] = ad
        h["ad_findings"] = ad.get("findings", [])
        # v6: credential-exposure findings + regulatory compliance cross-ref.
        h["weak_auth"] = compliance.weak_auth_findings(h)
        h["compliance"] = compliance.assess_host(h, weak=h["weak_auth"])
        hosts.append(h)

    def ip_key(h):
        try:
            return int(ipaddress.ip_address(h["ip"]))
        except Exception:
            return 0
    hosts.sort(key=ip_key)
    return hosts