admin / Synapse-NetscanXi

"""
NetScan Xi - patching engine (Patch and Remedy).

This module performs *active* remediation against assets that NetScan Xi has
scanned:

  * **Operating-system updates & upgrades over SSH** (``os_patch_host``) -
    apt / dnf / yum / zypper / apk are auto-detected on the target and an
    update + upgrade (or full distribution upgrade) is applied. A simulate mode
    runs the package manager in dry-run so nothing is changed.
  * **Docker image updates** (``patch_host``) - pulls the latest image for each
    running container over the same Docker Engine API the scanner uses, with an
    optional experimental container recreate.

Credential handling - IMPORTANT:
  SSH/sudo and registry credentials are supplied *per patch run* by the caller
  and live only for the duration of that call. They are **never written to the
  database, never logged, and never returned** to the client. Nothing here
  persists secrets.
"""

from __future__ import annotations

import base64
import json
import socket
import ssl
from typing import Any, Dict, List, Optional, Tuple
from urllib.error import URLError
from urllib.request import Request, urlopen

# Reuse the scanner's transport knobs so patching behaves exactly like
# inventory (same TLS posture, same endpoint discovery, same timeouts).
from . import docker_scan

# Pulls can take a while (large layers); give them more headroom than a GET.
PULL_TIMEOUT = 600.0
ACTION_TIMEOUT = 60.0


# ---------------------------------------------------------------------------
# Image reference helpers
# ---------------------------------------------------------------------------
def split_image_ref(ref: str) -> Tuple[str, str]:
    """Split an image reference into (fromImage, tag) for /images/create.

    Handles digests (``repo@sha256:...``), registry ports (``host:5000/x``) and
    bare names (``nginx`` -> tag ``latest``)."""
    ref = (ref or "").strip()
    if not ref:
        return "", ""
    if "@" in ref:
        # Digest-pinned: pull the whole ref as fromImage, no separate tag.
        return ref, ""
    # A ':' is only a tag separator if it's in the final path segment.
    last_slash = ref.rfind("/")
    last_colon = ref.rfind(":")
    if last_colon > last_slash:
        return ref[:last_colon], ref[last_colon + 1:]
    return ref, "latest"


def registry_auth_header(username: str, password: str,
                         server: str = "") -> Optional[str]:
    """Build a Docker ``X-Registry-Auth`` header value from ephemeral creds.

    Returns None when no username is supplied (public images need no auth).
    The returned string is base64(JSON) as required by the Engine API and is
    used once, then discarded by the caller."""
    if not username:
        return None
    payload = {"username": username, "password": password or ""}
    if server:
        payload["serveraddress"] = server
    raw = json.dumps(payload).encode("utf-8")
    return base64.b64encode(raw).decode("ascii")


# ---------------------------------------------------------------------------
# Engine API write transport (mirrors docker_scan's read transport)
# ---------------------------------------------------------------------------
def _post(base: str, path: str, headers: Optional[Dict[str, str]] = None,
          timeout: float = ACTION_TIMEOUT) -> Tuple[bool, str]:
    """POST to a Docker Engine API endpoint. Returns (ok, body_text)."""
    headers = dict(headers or {})
    try:
        if base.startswith("unix://"):
            return _post_unix(base[len("unix://"):], path, headers, timeout)
        url = base + path
        ctx = docker_scan._tls_context() if url.startswith("https") else None
        req = Request(url, data=b"", method="POST", headers=headers)
        with urlopen(req, timeout=timeout, context=ctx) as resp:
            body = resp.read().decode("utf-8", "replace")
            return 200 <= resp.status < 300, body
    except URLError as e:
        # HTTPError is a URLError subclass and carries the response body.
        body = ""
        try:
            body = e.read().decode("utf-8", "replace")  # type: ignore[attr-defined]
        except Exception:
            body = getattr(e, "reason", str(e)) or str(e)
        return False, body
    except (OSError, ValueError, ssl.SSLError) as e:
        return False, str(e)
    except Exception as e:  # never let a patch take the server down
        return False, str(e)


def _post_unix(sock_path: str, path: str, headers: Dict[str, str],
               timeout: float) -> Tuple[bool, str]:
    """Minimal HTTP-over-unix-socket POST for the local Docker daemon."""
    try:
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.settimeout(timeout)
        s.connect(sock_path)
        hdr_lines = [f"POST {path} HTTP/1.1", "Host: docker",
                     "Accept: application/json", "Content-Length: 0"]
        for k, v in headers.items():
            hdr_lines.append(f"{k}: {v}")
        hdr_lines.append("Connection: close")
        req = ("\r\n".join(hdr_lines) + "\r\n\r\n").encode()
        s.sendall(req)
        chunks = []
        while True:
            b = s.recv(65536)
            if not b:
                break
            chunks.append(b)
        s.close()
        raw = b"".join(chunks)
        head, _, body = raw.partition(b"\r\n\r\n")
        status_line = head.split(b"\r\n", 1)[0].decode("latin-1", "replace")
        ok = " 2" in status_line[:12]  # HTTP/1.1 2xx
        return ok, body.decode("utf-8", "replace")
    except Exception as e:
        return False, str(e)


def _stream_has_error(body: str) -> Optional[str]:
    """/images/create streams JSON lines; surface any error line."""
    for line in body.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            obj = json.loads(line)
        except ValueError:
            continue
        if isinstance(obj, dict) and (obj.get("error") or obj.get("errorDetail")):
            return obj.get("error") or json.dumps(obj.get("errorDetail"))
    return None


# ---------------------------------------------------------------------------
# Core operations
# ---------------------------------------------------------------------------
def pull_image(base: str, image_ref: str,
               auth_header: Optional[str] = None) -> Dict[str, Any]:
    """Pull (update) one image on a host's daemon. Best-effort, structured."""
    from_image, tag = split_image_ref(image_ref)
    if not from_image:
        return {"image": image_ref, "ok": False, "error": "unparseable image ref"}
    qs = f"/images/create?fromImage={from_image}"
    if tag:
        qs += f"&tag={tag}"
    headers = {"X-Registry-Auth": auth_header} if auth_header else {}
    ok, body = _post(base, qs, headers=headers, timeout=PULL_TIMEOUT)
    err = _stream_has_error(body) if ok else (body.strip()[:300] or "pull failed")
    if ok and not err:
        return {"image": image_ref, "ok": True}
    return {"image": image_ref, "ok": False, "error": err or "pull failed"}


def recreate_container(base: str, container: Dict[str, Any],
                       image_ref: str) -> Dict[str, Any]:
    """EXPERIMENTAL: recreate a container onto the freshly pulled image.

    Conservative flow: inspect -> stop -> remove -> create (same name/config) ->
    start. Defaulted OFF at the API layer. Returns a structured result."""
    cid = container.get("id") or ""
    name = container.get("name") or cid
    if not cid:
        return {"container": name, "ok": False, "error": "no container id"}

    # 1) Inspect to capture the existing Config + HostConfig.
    insp = docker_scan._api_get(base, f"/containers/{cid}/json")
    if not isinstance(insp, dict):
        return {"container": name, "ok": False, "error": "inspect failed"}
    cfg = dict(insp.get("Config") or {})
    cfg["Image"] = image_ref
    create_body = {k: v for k, v in cfg.items() if v is not None}
    create_body["HostConfig"] = insp.get("HostConfig") or {}
    nets = ((insp.get("NetworkSettings") or {}).get("Networks") or {})
    if nets:
        create_body["NetworkingConfig"] = {"EndpointsConfig": nets}

    # 2) Stop + remove the old container.
    _post(base, f"/containers/{cid}/stop", timeout=ACTION_TIMEOUT)
    ok_rm, body_rm = _post(base, f"/containers/{cid}?force=1",
                           timeout=ACTION_TIMEOUT)
    # DELETE is the correct verb; emulate via override header is messy, so use
    # the dedicated helper below instead.
    if not ok_rm:
        ok_rm, body_rm = _delete(base, f"/containers/{cid}?force=1")
    if not ok_rm:
        return {"container": name, "ok": False,
                "error": f"remove failed: {body_rm[:200]}"}

    # 3) Create with the same name + start.
    ok_c, body_c = _post_json(base, f"/containers/create?name={name}", create_body)
    if not ok_c:
        return {"container": name, "ok": False,
                "error": f"create failed: {body_c[:200]}"}
    try:
        new_id = (json.loads(body_c) or {}).get("Id", "")
    except ValueError:
        new_id = ""
    ok_s, body_s = _post(base, f"/containers/{new_id or name}/start",
                         timeout=ACTION_TIMEOUT)
    if not ok_s:
        return {"container": name, "ok": False,
                "error": f"start failed: {body_s[:200]}"}
    return {"container": name, "ok": True}


def _delete(base: str, path: str,
            timeout: float = ACTION_TIMEOUT) -> Tuple[bool, str]:
    try:
        if base.startswith("unix://"):
            return _post_unix(base[len("unix://"):], path, {"X-HTTP-Method": "DELETE"}, timeout)
        url = base + path
        ctx = docker_scan._tls_context() if url.startswith("https") else None
        req = Request(url, method="DELETE")
        with urlopen(req, timeout=timeout, context=ctx) as resp:
            return 200 <= resp.status < 300, resp.read().decode("utf-8", "replace")
    except URLError as e:
        try:
            return False, e.read().decode("utf-8", "replace")  # type: ignore[attr-defined]
        except Exception:
            return False, str(getattr(e, "reason", e))
    except Exception as e:
        return False, str(e)


def _post_json(base: str, path: str, payload: Dict[str, Any],
               timeout: float = ACTION_TIMEOUT) -> Tuple[bool, str]:
    data = json.dumps(payload).encode("utf-8")
    headers = {"Content-Type": "application/json"}
    try:
        if base.startswith("unix://"):
            return _post_json_unix(base[len("unix://"):], path, data, timeout)
        url = base + path
        ctx = docker_scan._tls_context() if url.startswith("https") else None
        req = Request(url, data=data, method="POST", headers=headers)
        with urlopen(req, timeout=timeout, context=ctx) as resp:
            return 200 <= resp.status < 300, resp.read().decode("utf-8", "replace")
    except URLError as e:
        try:
            return False, e.read().decode("utf-8", "replace")  # type: ignore[attr-defined]
        except Exception:
            return False, str(getattr(e, "reason", e))
    except Exception as e:
        return False, str(e)


def _post_json_unix(sock_path: str, path: str, data: bytes,
                    timeout: float) -> Tuple[bool, str]:
    try:
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        s.settimeout(timeout)
        s.connect(sock_path)
        head = (f"POST {path} HTTP/1.1\r\nHost: docker\r\n"
                f"Content-Type: application/json\r\n"
                f"Content-Length: {len(data)}\r\nConnection: close\r\n\r\n").encode()
        s.sendall(head + data)
        chunks = []
        while True:
            b = s.recv(65536)
            if not b:
                break
            chunks.append(b)
        s.close()
        raw = b"".join(chunks)
        head_b, _, body = raw.partition(b"\r\n\r\n")
        status_line = head_b.split(b"\r\n", 1)[0].decode("latin-1", "replace")
        ok = " 2" in status_line[:12]
        return ok, body.decode("utf-8", "replace")
    except Exception as e:
        return False, str(e)


# ---------------------------------------------------------------------------
# Host-level orchestration
# ---------------------------------------------------------------------------
def host_docker_endpoint(host: Dict[str, Any]) -> Optional[str]:
    """Resolve the reachable Docker API base URL for a host, if any.

    Prefer the endpoint the scanner already proved reachable; otherwise fall
    back to rebuilding candidates from the host's open docker ports."""
    d = host.get("docker") or {}
    api = d.get("api") or {}
    if api.get("reachable") and api.get("endpoint"):
        return api["endpoint"]
    ports = [str(p) for p in (d.get("ports") or [])]
    if not ports:
        # docker.ports may be under the detector block; try open_ports too.
        ports = [str(p.get("port")) for p in (host.get("port_details") or [])
                 if p.get("port")]
    eps = docker_scan._build_endpoints(host.get("ip", ""), ports)
    return eps[0] if eps else None


def plan_host(host: Dict[str, Any]) -> Dict[str, Any]:
    """Return what *would* be patched on a host (dry-run, no side effects)."""
    d = host.get("docker") or {}
    containers = [c for c in (d.get("containers") or [])
                  if c.get("state") == "running" and c.get("image")]
    images = sorted({c["image"] for c in containers})
    return {
        "asset_id": host.get("asset_id"),
        "ip": host.get("ip"),
        "endpoint": host_docker_endpoint(host),
        "images": images,
        "container_count": len(containers),
    }


def patch_host(host: Dict[str, Any],
               username: str = "", password: str = "", server: str = "",
               recreate: bool = False) -> Dict[str, Any]:
    """Patch one host's Docker images. Ephemeral creds; nothing persisted.

    Returns a structured per-image (and optionally per-container) result."""
    base = host_docker_endpoint(host)
    result: Dict[str, Any] = {
        "asset_id": host.get("asset_id"), "ip": host.get("ip"),
        "endpoint": base, "ok": False, "pulled": [], "recreated": [],
        "error": "",
    }
    if not base:
        result["error"] = "no reachable Docker API endpoint"
        return result

    # Build the one-shot auth header, then forget the raw secret immediately.
    auth_header = registry_auth_header(username, password, server)
    username = password = server = ""  # noqa: F841 - defensive scrub

    d = host.get("docker") or {}
    running = [c for c in (d.get("containers") or [])
               if c.get("state") == "running" and c.get("image")]
    if not running:
        result["error"] = "no running containers to update"
        return result

    # Pull each distinct image once.
    seen: Dict[str, Dict[str, Any]] = {}
    for c in running:
        img = c["image"]
        if img not in seen:
            seen[img] = pull_image(base, img, auth_header)
    result["pulled"] = list(seen.values())

    # Optionally recreate containers whose image pulled cleanly.
    if recreate:
        for c in running:
            pr = seen.get(c["image"], {})
            if pr.get("ok"):
                result["recreated"].append(recreate_container(base, c, c["image"]))

    pull_fail = [p for p in result["pulled"] if not p.get("ok")]
    rec_fail = [r for r in result["recreated"] if not r.get("ok")]
    result["ok"] = not pull_fail and not rec_fail
    if not result["ok"]:
        bits = []
        if pull_fail:
            bits.append(f"{len(pull_fail)} image pull(s) failed")
        if rec_fail:
            bits.append(f"{len(rec_fail)} container recreate(s) failed")
        result["error"] = "; ".join(bits)
    auth_header = None  # scrub
    return result


# ===========================================================================
# Operating-system patching over SSH (apt / dnf / yum / zypper / apk)
#
# Credentials (SSH + sudo) are supplied per run and are never stored or logged.
# paramiko is imported lazily so the rest of the app (and Docker patching) work
# even when it is not installed; ssh_available() reports its presence.
# ===========================================================================
SSH_TIMEOUT = 30.0
SSH_CMD_TIMEOUT = 1800.0  # OS upgrades can be slow
MAX_LOG_CHARS = 8000

# Package-manager command sets. Each maps to (refresh_cmd, upgrade_cmd,
# full_upgrade_cmd, simulate_flag_inserted_into_upgrade).
_PKG_MANAGERS = ("apt-get", "dnf", "yum", "zypper", "apk")

_OS_HINTS = (
    ("ubuntu", "apt-get"), ("debian", "apt-get"), ("mint", "apt-get"),
    ("raspbian", "apt-get"), ("kali", "apt-get"),
    ("fedora", "dnf"), ("red hat", "dnf"), ("redhat", "dnf"), ("rhel", "dnf"),
    ("centos", "yum"), ("rocky", "dnf"), ("almalinux", "dnf"), ("oracle", "dnf"),
    ("suse", "zypper"), ("sles", "zypper"),
    ("alpine", "apk"),
)


def ssh_available() -> bool:
    """True when paramiko (the SSH client) is importable."""
    try:
        import paramiko  # noqa: F401
        return True
    except Exception:
        return False


def pkg_manager_hint(os_string: str) -> str:
    """Best-effort package-manager guess from the scanner's OS string."""
    s = (os_string or "").lower()
    for needle, mgr in _OS_HINTS:
        if needle in s:
            return mgr
    if "windows" in s:
        return "windows (unsupported by SSH OS patching)"
    return ""


def os_plan_host(host: Dict[str, Any]) -> Dict[str, Any]:
    """Dry-run view of an OS patch target - no SSH, no side effects."""
    return {
        "asset_id": host.get("asset_id"),
        "ip": host.get("ip"),
        "hostname": (host.get("hostname") or "").replace("-", "") and host.get("hostname") or "",
        "os": host.get("os") or "unknown",
        "pkg_hint": pkg_manager_hint(host.get("os") or ""),
    }


def _commands_for(manager: str, dist_upgrade: bool, simulate: bool) -> List[str]:
    """Return the ordered shell commands to run for a manager.

    All privileged commands use ``sudo -S -p ''`` so the sudo password can be
    fed on stdin without echoing a prompt. Simulate mode never changes the box.
    """
    s = simulate
    if manager == "apt-get":
        up = "full-upgrade" if dist_upgrade else "upgrade"
        sim = " -s" if s else ""
        return [
            "sudo -S -p '' apt-get update",
            f"sudo -S -p '' DEBIAN_FRONTEND=noninteractive apt-get -y{sim} {up}",
        ]
    if manager in ("dnf", "yum"):
        if s:
            return [f"sudo -S -p '' {manager} -y --setopt=tsflags=test upgrade"]
        return [f"sudo -S -p '' {manager} -y upgrade"]
    if manager == "zypper":
        sim = " --dry-run" if s else ""
        sub = "dist-upgrade" if dist_upgrade else "update"
        return [
            "sudo -S -p '' zypper --non-interactive refresh",
            f"sudo -S -p '' zypper --non-interactive{sim} {sub}",
        ]
    if manager == "apk":
        sim = " -s" if s else ""
        return [
            "sudo -S -p '' apk update",
            f"sudo -S -p '' apk{sim} upgrade",
        ]
    return []


def _count_updates(manager: str, output: str) -> Optional[int]:
    """Best-effort count of packages changed, parsed from manager output."""
    import re
    try:
        if manager == "apt-get":
            m = re.search(r"(\d+)\s+upgraded,\s+(\d+)\s+newly installed", output)
            if m:
                return int(m.group(1)) + int(m.group(2))
        if manager in ("dnf", "yum"):
            n = len(re.findall(r"(?m)^\s*(?:Upgrading|Installing|Updating)\s", output))
            if n:
                return n
        if manager == "apk":
            n = len(re.findall(r"(?m)^\(\d+/\d+\)\sUpgrading", output))
            if n:
                return n
    except Exception:
        return None
    return None


def os_patch_host(host: Dict[str, Any], username: str, password: str = "",
                  port: int = 22, sudo_password: str = "",
                  dist_upgrade: bool = False, simulate: bool = False) -> Dict[str, Any]:
    """Apply OS updates/upgrades to one host over SSH. Ephemeral creds.

    Returns a structured result with the package manager, an update count when
    parseable, a (truncated) combined log, and ok/error."""
    ip = host.get("ip") or ""
    result: Dict[str, Any] = {
        "asset_id": host.get("asset_id"), "ip": ip, "ok": False,
        "manager": "", "updated_count": None, "log": "", "error": "",
        "simulated": bool(simulate),
    }
    if not ip:
        result["error"] = "asset has no IP address"
        return result
    try:
        import paramiko
    except Exception:
        result["error"] = "SSH support (paramiko) is not installed on the NetscanXi host"
        return result

    # sudo uses its own password if supplied, otherwise the login password.
    eff_sudo = sudo_password or password
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    log_parts: List[str] = []
    try:
        client.connect(hostname=ip, port=int(port or 22), username=username,
                       password=password or None, timeout=SSH_TIMEOUT,
                       allow_agent=False, look_for_keys=False)
    except Exception as e:
        result["error"] = f"SSH connect failed: {e}"
        password = sudo_password = eff_sudo = ""  # scrub
        try:
            client.close()
        except Exception:
            pass
        return result

    def _run(cmd: str) -> Tuple[int, str]:
        stdin, stdout, stderr = client.exec_command(cmd, timeout=SSH_CMD_TIMEOUT)
        if "sudo -S" in cmd:
            try:
                stdin.write((eff_sudo or "") + "\n")
                stdin.flush()
            except Exception:
                pass
        out = stdout.read().decode("utf-8", "replace")
        err = stderr.read().decode("utf-8", "replace")
        rc = stdout.channel.recv_exit_status()
        # Drop sudo's own password-prompt noise from the log.
        combined = (out + ("\n" + err if err.strip() else "")).strip()
        return rc, combined

    try:
        # Detect the package manager on the box.
        mgr = ""
        rc, which_out = _run("for m in apt-get dnf yum zypper apk; do "
                             "command -v $m >/dev/null 2>&1 && echo $m && break; done")
        cand = which_out.strip().splitlines()[-1].strip() if which_out.strip() else ""
        if cand in _PKG_MANAGERS:
            mgr = cand
        if not mgr:
            result["error"] = "no supported package manager (apt/dnf/yum/zypper/apk) found"
            return result
        result["manager"] = mgr

        cmds = _commands_for(mgr, dist_upgrade, simulate)
        final_rc = 0
        for c in cmds:
            rc, out = _run(c)
            # Show the command (with the sudo password flag, never the secret).
            log_parts.append(f"$ {c}\n{out}")
            final_rc = rc  # success is judged on the last (upgrade) command
        result["log"] = ("\n\n".join(log_parts))[-MAX_LOG_CHARS:]
        result["updated_count"] = _count_updates(mgr, "\n".join(log_parts))
        result["ok"] = (final_rc == 0)
        if not result["ok"]:
            result["error"] = f"{mgr} exited with code {final_rc}"
    except Exception as e:
        result["error"] = f"patch run failed: {e}"
        result["log"] = ("\n\n".join(log_parts))[-MAX_LOG_CHARS:]
    finally:
        password = sudo_password = eff_sudo = ""  # scrub
        try:
            client.close()
        except Exception:
            pass
    return result