admin / Synapse-NetscanXi

#!/usr/bin/env python3
"""
NetscanXi Version 10r1 - Docker / container scanning & vulnerability assessment.

This module extends NetscanXi's network-level Docker *detection* (parser.py)
into deep, per-container/-image visibility:

  1. Authenticated Docker Engine API inventory
        - GET /version, /info     -> engine version, storage driver, rootless,
                                       live-restore, registry config
        - GET /containers/json    -> running/stopped containers (image, ports,
                                       privileged, mounts, restart policy)
        - GET /images/json        -> image inventory (tags, digests, size)
     Works over plain HTTP (2375), TLS (2376, with optional client cert) or a
     local unix socket.

  2. Image vulnerability scanning
        - Shells out to a locally-installed scanner (Trivy or Grype) per image
          ref/digest and parses OS-package + language CVEs. Fully offline once
          the scanner's DB is populated. Results are normalised to the same CVE
          shape used elsewhere so enrich.py (KEV/EPSS) and risk scoring apply.

  3. CIS Docker Benchmark misconfiguration audit
        - Engine + per-container checks (exposed API, privileged, host net/pid,
          docker.sock mounts, root user, :latest, restart policy, ...), each a
          pass / warn / fail finding mapped to CIS Docker controls.

  4. Registry detection + SBOM
        - Flags exposed/insecure registries (5000) and, when the scanner
          supports it, can emit an SBOM (CycloneDX/SPDX) per image.

  5. Orchestration awareness
        - Surfaces Swarm / Kubernetes indicators already gathered by the port
          probe (2377/7946 swarm, 6443/2379/10250 k8s) as structured findings.

EVERYTHING is best-effort and degrades gracefully: if the Docker API is not
reachable and no scanner is installed, the module still returns the original
network-level detection plus a clear "capabilities" block describing what was
and wasn't available. No feature is *required* for the rest of NetscanXi to
work, and nothing here ever runs an exploit or mutates the target.
"""

from __future__ import annotations

import json
import os
import shutil
import socket
import ssl
import subprocess
import time
from typing import Any, Dict, List, Optional, Tuple
from urllib.request import urlopen, Request
from urllib.error import URLError

# ---------------------------------------------------------------------------
# Configuration (all overridable via environment, all optional)
# ---------------------------------------------------------------------------
# Whether to attempt authenticated Docker API inventory at all.
DOCKER_API_ENABLED = os.environ.get("NETSCAN_DOCKER_API", "1") not in ("0", "false", "False")
# Whether to run an image vulnerability scanner (trivy/grype) when present.
DOCKER_IMAGE_SCAN = os.environ.get("NETSCAN_DOCKER_IMAGE_SCAN", "1") not in ("0", "false", "False")
# Preferred scanner: "auto" picks trivy then grype; or force "trivy"/"grype".
DOCKER_SCANNER = os.environ.get("NETSCAN_DOCKER_SCANNER", "auto").lower()
# Per-API-call timeout (seconds) and per-image scan timeout.
API_TIMEOUT = float(os.environ.get("NETSCAN_DOCKER_API_TIMEOUT", "6"))
SCAN_TIMEOUT = float(os.environ.get("NETSCAN_DOCKER_SCAN_TIMEOUT", "180"))
# Max images to scan per host (cost guard).
MAX_IMAGES = int(os.environ.get("NETSCAN_DOCKER_MAX_IMAGES", "25"))

# Optional TLS client materials for 2376 (mTLS) and optional bearer creds.
DOCKER_TLS_CA = os.environ.get("NETSCAN_DOCKER_TLS_CA", "")
DOCKER_TLS_CERT = os.environ.get("NETSCAN_DOCKER_TLS_CERT", "")
DOCKER_TLS_KEY = os.environ.get("NETSCAN_DOCKER_TLS_KEY", "")


# ===========================================================================
# Scanner discovery
# ===========================================================================
def _find_scanner() -> Tuple[Optional[str], Optional[str]]:
    """Return (name, path) of an available image scanner, or (None, None)."""
    order = ["trivy", "grype"]
    if DOCKER_SCANNER in ("trivy", "grype"):
        order = [DOCKER_SCANNER]
    for name in order:
        path = shutil.which(name)
        if path:
            return name, path
    return None, None


def scanner_info() -> Dict[str, Any]:
    """Describe the available image scanner (for the capabilities block)."""
    name, path = _find_scanner()
    return {
        "available": bool(path),
        "name": name or "",
        "path": path or "",
        "sbom": bool(path),   # both trivy and grype/syft can emit an SBOM
    }


# ===========================================================================
# Docker Engine API client (HTTP / HTTPS / unix socket)
# ===========================================================================
def _build_endpoints(host: str, ports: List[str]) -> List[str]:
    """Candidate base URLs to try for a host, in order of preference."""
    eps = []
    p = [str(x) for x in (ports or [])]
    if "2376" in p:
        eps.append(f"https://{host}:2376")
    if "2375" in p:
        eps.append(f"http://{host}:2375")
    if "4243" in p:
        eps.append(f"http://{host}:4243")
    # Local socket as a last resort when scanning localhost.
    if host in ("127.0.0.1", "localhost", "::1") and os.path.exists("/var/run/docker.sock"):
        eps.append("unix:///var/run/docker.sock")
    return eps


def _tls_context() -> ssl.SSLContext:
    ctx = ssl.create_default_context()
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE  # hosts often use self-signed daemon certs
    if DOCKER_TLS_CA and os.path.exists(DOCKER_TLS_CA):
        try:
            ctx.load_verify_locations(DOCKER_TLS_CA)
            ctx.verify_mode = ssl.CERT_REQUIRED
            ctx.check_hostname = False
        except Exception:
            pass
    if DOCKER_TLS_CERT and DOCKER_TLS_KEY and os.path.exists(DOCKER_TLS_CERT):
        try:
            ctx.load_cert_chain(DOCKER_TLS_CERT, DOCKER_TLS_KEY)
        except Exception:
            pass
    return ctx


def _api_get(base: str, path: str) -> Optional[Any]:
    """GET a Docker API endpoint, returning decoded JSON or None on failure."""
    try:
        if base.startswith("unix://"):
            return _api_get_unix(base[len("unix://"):], path)
        url = base + path
        ctx = _tls_context() if url.startswith("https") else None
        req = Request(url, headers={"Accept": "application/json"})
        with urlopen(req, timeout=API_TIMEOUT, context=ctx) as resp:
            data = resp.read()
        return json.loads(data.decode("utf-8", "replace")) if data else None
    except (URLError, OSError, ValueError, ssl.SSLError):
        return None
    except Exception:
        return None


def _api_get_unix(sock_path: str, path: str) -> Optional[Any]:
    """Minimal HTTP-over-unix-socket GET for the local Docker daemon."""
    try:
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.settimeout(API_TIMEOUT)
        s.connect(sock_path)
        req = f"GET {path} HTTP/1.1\r\nHost: docker\r\nAccept: application/json\r\nConnection: close\r\n\r\n"
        s.sendall(req.encode())
        chunks = []
        while True:
            b = s.recv(65536)
            if not b:
                break
            chunks.append(b)
        s.close()
        raw = b"".join(chunks)
        if b"\r\n\r\n" not in raw:
            return None
        body = raw.split(b"\r\n\r\n", 1)[1]
        # Handle chunked transfer encoding crudely: find the JSON payload.
        text = body.decode("utf-8", "replace").strip()
        start = text.find("[")
        brace = text.find("{")
        if start < 0 or (0 <= brace < start):
            start = brace
        if start < 0:
            return None
        end = max(text.rfind("]"), text.rfind("}"))
        try:
            return json.loads(text[start:end + 1])
        except ValueError:
            return None
    except Exception:
        return None


def api_inventory(host: str, ports: List[str]) -> Dict[str, Any]:
    """
    Pull engine/version/info + containers + images from the first reachable
    Docker API endpoint. Returns a dict with 'reachable' False if none answered.
    """
    result: Dict[str, Any] = {
        "reachable": False, "endpoint": "", "secure": None,
        "engine": {}, "containers": [], "images": [],
    }
    if not DOCKER_API_ENABLED:
        result["reason"] = "api disabled by configuration"
        return result

    for base in _build_endpoints(host, ports):
        ver = _api_get(base, "/version")
        if ver is None:
            continue
        info = _api_get(base, "/info") or {}
        containers_raw = _api_get(base, "/containers/json?all=1") or []
        images_raw = _api_get(base, "/images/json") or []

        result["reachable"] = True
        result["endpoint"] = base
        result["secure"] = base.startswith("https") or base.startswith("unix")
        result["engine"] = _shape_engine(ver, info)
        containers = [_shape_container(c) for c in containers_raw]
        # Backfill IPs (when the list omitted them) and authoritative health
        # detail via inspect for running containers.
        for shaped, raw in zip(containers, containers_raw):
            if shaped.get("state") != "running":
                continue
            need_ip = not shaped.get("ips")
            need_health = shaped.get("health") != "none"
            if need_ip or need_health:
                _backfill_container_detail(base, raw.get("Id", ""), shaped,
                                           want_ip=need_ip, want_health=need_health)
        result["containers"] = containers
        result["images"] = [_shape_image(i) for i in images_raw]
        return result

    result["reason"] = "no Docker API endpoint answered"
    return result


def _backfill_container_detail(base: str, cid: str, shaped: Dict[str, Any],
                              want_ip: bool = True, want_health: bool = True) -> None:
    """Fill container IPs/networks and authoritative health from
    /containers/{id}/json. The list endpoint omits IPs on some daemon/network
    drivers, and only encodes health as a substring; inspect gives
    State.Health.{Status,FailingStreak,Log}."""
    if not cid:
        return
    detail = _api_get(base, f"/containers/{cid}/json")
    if not isinstance(detail, dict):
        return

    # --- Authoritative health (State.Health) ---
    if want_health:
        health = ((detail.get("State") or {}).get("Health") or {})
        hstatus = (health.get("Status") or "").lower()
        if hstatus:
            shaped["health"] = hstatus  # healthy / unhealthy / starting
            shaped["health_failing_streak"] = int(health.get("FailingStreak") or 0)
            log = health.get("Log") or []
            if log and isinstance(log[-1], dict):
                shaped["health_last_output"] = (log[-1].get("Output") or "").strip()[:200]
        elif shaped.get("health") == "none":
            shaped["health"] = "none"

    if not want_ip:
        return
    ns = (detail.get("NetworkSettings") or {})
    nets = ns.get("Networks") or {}
    networks, ips = [], []
    for net_name, net in nets.items():
        if not isinstance(net, dict):
            continue
        ip = net.get("IPAddress") or ""
        ip6 = net.get("GlobalIPv6Address") or ""
        networks.append({
            "network": net_name, "ip": ip, "ipv6": ip6,
            "mac": net.get("MacAddress", ""),
            "gateway": net.get("Gateway", ""),
            "aliases": net.get("Aliases") or [],
        })
        if ip:
            ips.append(ip)
        if ip6:
            ips.append(ip6)
    # Fall back to the top-level IPAddress (default bridge) if still nothing.
    if not ips and ns.get("IPAddress"):
        ips.append(ns["IPAddress"])
    if networks:
        shaped["networks"] = networks
    if ips:
        shaped["ips"] = ips


def _shape_engine(ver: Dict[str, Any], info: Dict[str, Any]) -> Dict[str, Any]:
    rootless = False
    sec = info.get("SecurityOptions") or []
    if any("rootless" in str(s).lower() for s in sec):
        rootless = True
    return {
        "version": ver.get("Version", ""),
        "api_version": ver.get("ApiVersion", ""),
        "os": ver.get("Os", ""),
        "arch": ver.get("Arch", ""),
        "kernel": ver.get("KernelVersion", ""),
        "storage_driver": info.get("Driver", ""),
        "containers": info.get("Containers", 0),
        "containers_running": info.get("ContainersRunning", 0),
        "images": info.get("Images", 0),
        "rootless": rootless,
        "live_restore": bool(info.get("LiveRestoreEnabled")),
        "cgroup_version": info.get("CgroupVersion", ""),
        "experimental": bool(info.get("ExperimentalBuild")),
        "registry_insecure": [
            r.get("Name") for r in (
                (info.get("RegistryConfig") or {}).get("IndexConfigs") or {}
            ).values() if isinstance(r, dict) and r.get("Secure") is False
        ],
    }


def _parse_health(status: str) -> str:
    """Derive container health from the /containers/json Status string.
    Docker appends '(healthy)', '(unhealthy)' or '(health: starting)' when a
    HEALTHCHECK is defined. No marker => no healthcheck configured ('none').
    """
    s = (status or "").lower()
    if "(healthy)" in s:
        return "healthy"
    if "(unhealthy)" in s:
        return "unhealthy"
    if "health: starting" in s or "(starting)" in s:
        return "starting"
    return "none"


def _shape_container(c: Dict[str, Any]) -> Dict[str, Any]:
    names = c.get("Names") or []
    name = (names[0].lstrip("/") if names else c.get("Id", "")[:12])
    host_cfg = c.get("HostConfig") or {}
    mounts = c.get("Mounts") or []
    sock_mounted = any(
        "docker.sock" in str(m.get("Source", "")) for m in mounts
    )

    # Published (host->container) port maps AND internal-only exposed ports.
    ports = []
    published, internal = [], []
    for p in c.get("Ports") or []:
        if p.get("PublicPort"):
            mapping = f"{p.get('IP','0.0.0.0')}:{p['PublicPort']}->{p.get('PrivatePort')}/{p.get('Type','tcp')}"
            ports.append(mapping)
            published.append(mapping)
        elif p.get("PrivatePort"):
            ep = f"{p['PrivatePort']}/{p.get('Type','tcp')}"
            ports.append(ep)
            internal.append(ep)

    # Per-network IP addresses (NetworkSettings.Networks[].IPAddress).
    # Present in /containers/json for running containers; the dict is keyed
    # by docker network name (bridge, host, custom user nets, ...).
    networks = []
    ips = []
    ns = (c.get("NetworkSettings") or {}).get("Networks") or {}
    for net_name, net in ns.items():
        if not isinstance(net, dict):
            continue
        ip = net.get("IPAddress") or ""
        ip6 = net.get("GlobalIPv6Address") or ""
        networks.append({
            "network": net_name,
            "ip": ip,
            "ipv6": ip6,
            "mac": net.get("MacAddress", ""),
            "gateway": net.get("Gateway", ""),
            "aliases": net.get("Aliases") or [],
        })
        if ip:
            ips.append(ip)
        if ip6:
            ips.append(ip6)

    return {
        "id": (c.get("Id", "")[:12]),
        "name": name,
        "image": c.get("Image", ""),
        "image_id": c.get("ImageID", ""),
        "state": c.get("State", ""),
        "status": c.get("Status", ""),
        "health": _parse_health(c.get("Status", "")),
        "health_failing_streak": 0,
        "health_last_output": "",
        "ports": ports,
        "ports_published": published,
        "ports_internal": internal,
        "ips": ips,
        "networks": networks,
        "privileged": bool(host_cfg.get("Privileged")),
        "network_mode": host_cfg.get("NetworkMode", ""),
        "pid_mode": host_cfg.get("PidMode", ""),
        "restart_policy": (host_cfg.get("RestartPolicy") or {}).get("Name", ""),
        "cap_add": host_cfg.get("CapAdd") or [],
        "docker_sock_mounted": sock_mounted,
    }


def _shape_image(i: Dict[str, Any]) -> Dict[str, Any]:
    tags = i.get("RepoTags") or []
    digests = i.get("RepoDigests") or []
    tag = next((t for t in tags if t and t != "<none>:<none>"), "")
    return {
        "id": (i.get("Id", "")[:19]),
        "tags": [t for t in tags if t != "<none>:<none>"],
        "primary_tag": tag,
        "digest": digests[0] if digests else "",
        "size_mb": round((i.get("Size", 0) or 0) / 1_000_000, 1),
        "created": i.get("Created", 0),
    }


# ===========================================================================
# Image vulnerability scanning (Trivy / Grype)
# ===========================================================================
def scan_image(ref: str, scanner: Optional[Tuple[str, str]] = None) -> Dict[str, Any]:
    """
    Scan one image ref with the available scanner. Returns
    {ok, scanner, ref, vulns[], severity{}, error}. Each vuln is normalised to
    {id, cvss, severity, source, pkg, installed, fixed, title}.
    """
    name, path = scanner or _find_scanner()
    out: Dict[str, Any] = {"ok": False, "scanner": name or "", "ref": ref,
                           "vulns": [], "severity": {}, "error": ""}
    if not path:
        out["error"] = "no image scanner installed"
        return out
    if not ref or ref.startswith("<none>"):
        out["error"] = "unscannable image ref"
        return out
    try:
        if name == "trivy":
            cmd = [path, "image", "--quiet", "--format", "json",
                   "--scanners", "vuln", ref]
        else:  # grype
            cmd = [path, "-q", "-o", "json", ref]
        proc = subprocess.run(cmd, capture_output=True, text=True,
                              timeout=SCAN_TIMEOUT)
        if proc.returncode not in (0,) and not proc.stdout.strip():
            out["error"] = (proc.stderr or "scanner failed").strip()[:300]
            return out
        data = json.loads(proc.stdout)
        out["vulns"] = (_parse_trivy(data) if name == "trivy"
                        else _parse_grype(data))
        out["ok"] = True
    except subprocess.TimeoutExpired:
        out["error"] = "scan timed out"
    except (ValueError, OSError) as exc:
        out["error"] = str(exc)[:300]
    out["severity"] = _severity_counts(out["vulns"])
    return out


def _parse_trivy(data: Dict[str, Any]) -> List[Dict[str, Any]]:
    vulns = []
    for res in data.get("Results") or []:
        for v in res.get("Vulnerabilities") or []:
            cvss = 0.0
            cvd = v.get("CVSS") or {}
            for vendor in ("nvd", "redhat", "ghsa"):
                node = cvd.get(vendor) or {}
                cvss = node.get("V3Score") or node.get("V2Score") or cvss
                if cvss:
                    break
            vulns.append({
                "id": v.get("VulnerabilityID", ""),
                "cvss": str(cvss or ""),
                "severity": (v.get("Severity") or "").title(),
                "source": "trivy",
                "pkg": v.get("PkgName", ""),
                "installed": v.get("InstalledVersion", ""),
                "fixed": v.get("FixedVersion", ""),
                "title": (v.get("Title") or v.get("Description") or "")[:160],
            })
    return _dedupe(vulns)


def _parse_grype(data: Dict[str, Any]) -> List[Dict[str, Any]]:
    vulns = []
    for m in data.get("matches") or []:
        v = m.get("vulnerability") or {}
        art = m.get("artifact") or {}
        cvss = 0.0
        for c in v.get("cvss") or []:
            metrics = c.get("metrics") or {}
            cvss = metrics.get("baseScore") or cvss
            if cvss:
                break
        fix = (v.get("fix") or {}).get("versions") or []
        vulns.append({
            "id": v.get("id", ""),
            "cvss": str(cvss or ""),
            "severity": (v.get("severity") or "").title(),
            "source": "grype",
            "pkg": art.get("name", ""),
            "installed": art.get("version", ""),
            "fixed": ", ".join(fix),
            "title": (v.get("description") or "")[:160],
        })
    return _dedupe(vulns)


def _dedupe(vulns: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    seen, out = set(), []
    for v in vulns:
        key = (v.get("id"), v.get("pkg"))
        if v.get("id") and key not in seen:
            seen.add(key)
            out.append(v)
    return out


def _severity_counts(vulns: List[Dict[str, Any]]) -> Dict[str, int]:
    counts = {"Critical": 0, "High": 0, "Medium": 0, "Low": 0, "Unknown": 0}
    for v in vulns:
        s = (v.get("severity") or "Unknown").title()
        counts[s] = counts.get(s, 0) + 1
    return counts


# ===========================================================================
# CIS Docker Benchmark - misconfiguration audit
# ===========================================================================
def cis_audit(detection: Dict[str, Any], inv: Dict[str, Any]) -> List[Dict[str, Any]]:
    """
    Produce pass/warn/fail findings against high-signal CIS Docker controls,
    using whatever data we have (network detection + API inventory).
    """
    f: List[Dict[str, Any]] = []

    def add(status, check, detail, ref):
        f.append({"status": status, "check": check, "detail": detail, "ref": ref})

    # --- Engine / daemon ---
    if detection.get("api_exposed"):
        add("fail", "Docker API exposure",
            "Docker Engine API reachable over the network (2375/2376/4243); "
            "a remote API grants full control of the host.",
            "CIS Docker 2.x / 3.x")
    else:
        add("pass", "Docker API exposure",
            "No remotely reachable Docker API detected.", "CIS Docker 2.x")

    eng = inv.get("engine") or {}
    if inv.get("reachable"):
        if inv.get("secure") is False:
            add("fail", "Daemon TLS",
                "Docker API answered over plain HTTP (no TLS / client-cert auth).",
                "CIS Docker 2.4")
        else:
            add("pass", "Daemon TLS", "API endpoint uses TLS / local socket.",
                "CIS Docker 2.4")
        if eng.get("rootless"):
            add("pass", "Rootless mode", "Engine is running rootless.",
                "CIS Docker 2.x")
        else:
            add("warn", "Rootless mode",
                "Engine running as root (rootless mode not detected).",
                "CIS Docker 2.x")
        if eng.get("live_restore"):
            add("pass", "Live restore", "live-restore enabled.", "CIS Docker 2.14")
        else:
            add("warn", "Live restore",
                "live-restore disabled - containers stop if the daemon restarts.",
                "CIS Docker 2.14")
        for reg in eng.get("registry_insecure") or []:
            add("fail", "Insecure registry",
                f"Insecure (non-TLS) registry configured: {reg}", "CIS Docker 2.x")

    # --- Per-container ---
    for c in inv.get("containers") or []:
        cn = c.get("name", "?")
        if c.get("privileged"):
            add("fail", "Privileged container",
                f"Container '{cn}' runs with --privileged (full host capabilities).",
                "CIS Docker 5.4")
        if c.get("docker_sock_mounted"):
            add("fail", "Docker socket mount",
                f"Container '{cn}' mounts /var/run/docker.sock (container-escape path).",
                "CIS Docker 5.31")
        if str(c.get("network_mode", "")).lower() == "host":
            add("warn", "Host network mode",
                f"Container '{cn}' uses host networking (no network isolation).",
                "CIS Docker 5.29")
        if "host" in str(c.get("pid_mode", "")).lower():
            add("warn", "Host PID namespace",
                f"Container '{cn}' shares the host PID namespace.", "CIS Docker 5.15")
        if c.get("cap_add"):
            add("warn", "Added capabilities",
                f"Container '{cn}' adds capabilities: {', '.join(c['cap_add'])}.",
                "CIS Docker 5.3")
        if not c.get("restart_policy"):
            add("warn", "Restart policy",
                f"Container '{cn}' has no restart policy set.", "CIS Docker 5.14")
        # Runtime health (CIS Docker 5.26: check container health at runtime).
        health = c.get("health", "none")
        if c.get("state") == "running":
            if health == "unhealthy":
                streak = c.get("health_failing_streak") or 0
                add("fail", "Container health",
                    f"Container '{cn}' reports UNHEALTHY"
                    + (f" (failing streak {streak})" if streak else "") + ".",
                    "CIS Docker 5.26")
            elif health == "none":
                add("warn", "Container health",
                    f"Container '{cn}' has no HEALTHCHECK defined - health "
                    "cannot be monitored at runtime.", "CIS Docker 5.26")

    # --- Per-image ---
    for img in inv.get("images") or []:
        for t in img.get("tags") or []:
            if t.endswith(":latest"):
                add("warn", "Mutable image tag",
                    f"Image '{t}' uses the ':latest' tag (non-reproducible).",
                    "CIS Docker 4.x")
                break

    return f


# ===========================================================================
# Orchestration awareness
# ===========================================================================
_ORCH_PORTS = {
    "2377": "Docker Swarm management",
    "7946": "Docker Swarm gossip",
    "6443": "Kubernetes API server",
    "2379": "etcd (k8s/swarm control plane)",
    "2380": "etcd peer",
    "10250": "Kubelet API",
}


def orchestration(port_details: List[Dict[str, Any]]) -> Dict[str, Any]:
    found = []
    for pd in port_details or []:
        p = str(pd.get("port", ""))
        if p in _ORCH_PORTS:
            risky = p in ("10250", "6443", "2379")
            found.append({"port": p, "role": _ORCH_PORTS[p], "sensitive": risky})
    kind = ""
    pset = {x["port"] for x in found}
    if {"2377", "7946"} & pset:
        kind = "Docker Swarm"
    if {"6443", "10250"} & pset:
        kind = "Kubernetes" if not kind else "Swarm + Kubernetes"
    return {"detected": bool(found), "kind": kind, "indicators": found}


# ===========================================================================
# Top-level: enrich a single host's docker block
# ===========================================================================
def enrich_host(host: Dict[str, Any], scanner=None,
                image_scan: Optional[bool] = None) -> Dict[str, Any]:
    """
    Given a parsed host, deepen its 'docker' block in place with API inventory,
    image vulnerabilities, a CIS audit, orchestration findings and a summary.
    Safe to call on any host; no-ops cleanly when the host isn't Docker.
    Returns the (mutated) host.
    """
    det = host.get("docker") or {}
    if not det.get("is_docker"):
        return host

    do_image = DOCKER_IMAGE_SCAN if image_scan is None else image_scan
    sc = scanner if scanner is not None else _find_scanner()

    ip = host.get("ip", "")
    inv = api_inventory(ip, det.get("ports") or [])

    # Merge engine version discovered via API (more authoritative than NSE).
    if inv.get("engine", {}).get("version") and not det.get("version"):
        det["version"] = inv["engine"]["version"]

    # Image vulnerability scanning.
    image_reports = []
    img_sev_total = {"Critical": 0, "High": 0, "Medium": 0, "Low": 0, "Unknown": 0}
    total_image_vulns = 0
    if do_image and sc and sc[1] and inv.get("images"):
        refs = []
        for img in inv["images"]:
            ref = img.get("primary_tag") or img.get("digest") or ""
            if ref:
                refs.append((img, ref))
        for img, ref in refs[:MAX_IMAGES]:
            rep = scan_image(ref, scanner=sc)
            sev = rep.get("severity", {})
            for k, v in sev.items():
                img_sev_total[k] = img_sev_total.get(k, 0) + v
            total_image_vulns += len(rep.get("vulns", []))
            image_reports.append({
                "image": ref, "digest": img.get("digest", ""),
                "ok": rep.get("ok"), "error": rep.get("error", ""),
                "vuln_count": len(rep.get("vulns", [])),
                "severity": sev, "vulns": rep.get("vulns", []),
            })

    # CIS audit + orchestration.
    cis = cis_audit(det, inv)
    orch = orchestration(host.get("port_details") or [])

    det.update({
        "api": {
            "reachable": inv.get("reachable", False),
            "endpoint": inv.get("endpoint", ""),
            "secure": inv.get("secure"),
            "reason": inv.get("reason", ""),
        },
        "engine": inv.get("engine", {}),
        "containers": inv.get("containers", []),
        "images": inv.get("images", []),
        "image_scan": {
            "scanner": (sc[0] if sc else ""),
            "ran": bool(image_reports),
            "image_count": len(inv.get("images", [])),
            "scanned": len(image_reports),
            "total_vulns": total_image_vulns,
            "severity": img_sev_total,
            "reports": image_reports,
        },
        "cis_audit": cis,
        "cis_summary": {
            "pass": sum(1 for x in cis if x["status"] == "pass"),
            "warn": sum(1 for x in cis if x["status"] == "warn"),
            "fail": sum(1 for x in cis if x["status"] == "fail"),
        },
        "orchestration": orch,
        "capabilities": {
            "api_inventory": inv.get("reachable", False),
            "image_scanner": (sc[0] if sc and sc[1] else ""),
            "image_scanning": bool(image_reports),
            "sbom": bool(sc and sc[1]),
        },
        "summary": {
            "containers": len(inv.get("containers", [])),
            "containers_running": sum(
                1 for c in inv.get("containers", []) if c.get("state") == "running"),
            "containers_healthy": sum(
                1 for c in inv.get("containers", []) if c.get("health") == "healthy"),
            "containers_unhealthy": sum(
                1 for c in inv.get("containers", []) if c.get("health") == "unhealthy"),
            "containers_starting": sum(
                1 for c in inv.get("containers", []) if c.get("health") == "starting"),
            "containers_no_healthcheck": sum(
                1 for c in inv.get("containers", [])
                if c.get("state") == "running" and c.get("health") == "none"),
            "container_ips": sorted({
                ip for c in inv.get("containers", []) for ip in (c.get("ips") or [])
            }),
            "images": len(inv.get("images", [])),
            "image_vulns": total_image_vulns,
            "image_critical": img_sev_total.get("Critical", 0),
            "image_high": img_sev_total.get("High", 0),
            "privileged": sum(
                1 for c in inv.get("containers", []) if c.get("privileged")),
            "sock_mounts": sum(
                1 for c in inv.get("containers", []) if c.get("docker_sock_mounted")),
            "cis_fail": sum(1 for x in cis if x["status"] == "fail"),
            "cis_warn": sum(1 for x in cis if x["status"] == "warn"),
        },
    })
    host["docker"] = det
    return host


def enrich_hosts(hosts: List[Dict[str, Any]], image_scan: Optional[bool] = None) -> List[Dict[str, Any]]:
    """Enrich every Docker host in a scan. Reuses one scanner lookup."""
    sc = _find_scanner()
    for h in hosts:
        try:
            enrich_host(h, scanner=sc, image_scan=image_scan)
        except Exception:
            # Never let docker enrichment break a scan.
            pass
    return hosts


def capabilities() -> Dict[str, Any]:
    """Report what Docker scanning capabilities this install currently has."""
    sc = scanner_info()
    return {
        "api_enabled": DOCKER_API_ENABLED,
        "image_scan_enabled": DOCKER_IMAGE_SCAN,
        "scanner": sc,
        "max_images": MAX_IMAGES,
    }


def summarize(hosts: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Fleet-wide Docker rollup for the dashboard."""
    docker_hosts = [h for h in hosts if (h.get("docker") or {}).get("is_docker")]
    containers = running = images = img_vulns = img_crit = img_high = 0
    privileged = sock = api_exposed = cis_fail = scanned_hosts = 0
    healthy = unhealthy = no_healthcheck = 0
    for h in docker_hosts:
        d = h["docker"]
        s = d.get("summary") or {}
        containers += s.get("containers", 0)
        running += s.get("containers_running", 0)
        healthy += s.get("containers_healthy", 0)
        unhealthy += s.get("containers_unhealthy", 0)
        no_healthcheck += s.get("containers_no_healthcheck", 0)
        images += s.get("images", 0)
        img_vulns += s.get("image_vulns", 0)
        img_crit += s.get("image_critical", 0)
        img_high += s.get("image_high", 0)
        privileged += s.get("privileged", 0)
        sock += s.get("sock_mounts", 0)
        cis_fail += s.get("cis_fail", 0)
        if d.get("api", {}).get("reachable"):
            api_exposed += 1 if d.get("api_exposed") else 0
        if d.get("image_scan", {}).get("ran"):
            scanned_hosts += 1
    return {
        "docker_hosts": len(docker_hosts),
        "containers": containers,
        "containers_running": running,
        "containers_healthy": healthy,
        "containers_unhealthy": unhealthy,
        "containers_no_healthcheck": no_healthcheck,
        "images": images,
        "image_vulns": img_vulns,
        "image_critical": img_crit,
        "image_high": img_high,
        "privileged_containers": privileged,
        "sock_mounts": sock,
        "api_exposed_hosts": api_exposed,
        "cis_failures": cis_fail,
        "scanned_hosts": scanned_hosts,
    }