#!/usr/bin/env python3
"""Network Configurator - a local configuration job runner for network engineers.

    python3 network-configurator.py

opens a small web UI on 127.0.0.1 and nothing else. Everything runs on YOUR
machine, against YOUR network, with YOUR credentials. ciscotools.dev never
touches a device, never sees a credential, and never proxies a connection:
the only thing it serves is this file, its docs, and a license check that
sends your API key and this app's version and nothing else.

Lite (free, no key needed)
  * Inventory CSV: hostname, ip, platform, group (extra columns become
    per-device variables).
  * Show-command runner, with per-device output logged to a folder.
  * Dry-run plan view: exactly what would be sent, to what, in what order.
  * SecureCRT session export from the inventory (zip of .ini + sessions.csv).

Pro (unlocked with your ciscotools.dev API key)
  * Config-mode push from a template with per-device variables.
  * Custom command sets.
  * Saved job bundles.
  * Before/after diff of show output around a change.
  * Scheduled runs.

Requirements: Python 3.9 or newer and the system `ssh` client. No pip, no
venv, no third-party packages. Tested on 3.9 and 3.13.

Safety rails, always on:
  * DRY_RUN defaults to true. A dry run opens no SSH connection at all.
  * A real config-mode push asks for confirmation before it runs.
  * Every real write appends one JSON line to the audit log, after it
    succeeds, never before.
  * Passwords are never logged, never written to the audit log, and never
    put on a command line.

Copyright (c) 2026 Brody Networks. Provided as-is, no warranty. You are
responsible for what you run against your own network.
"""

from __future__ import annotations

APP_NAME = "network-configurator"
APP_TITLE = "Network Configurator"
APP_VERSION = "1.0.0"

import argparse
import csv
import datetime
import difflib
import getpass
import io
import ipaddress
import json
import os
import re
import secrets
import shutil
import ssl
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import webbrowser
import zipfile
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

# --- BEGIN ED25519 SHARED CORE (verbatim from backend/services/ed25519_pure.py; sync with backend/scripts/sync_ed25519_core.py) ---
import hashlib

_P = 2 ** 255 - 19
_L = 2 ** 252 + 27742317777372353535851937790883648493


def _sha512(data):
    return hashlib.sha512(data).digest()


def _inv(x):
    return pow(x, _P - 2, _P)


_D = -121665 * _inv(121666) % _P
_SQRT_M1 = pow(2, (_P - 1) // 4, _P)


def _recover_x(y, sign_bit):
    if y >= _P:
        return None
    x2 = (y * y - 1) * _inv(_D * y * y + 1) % _P
    if x2 == 0:
        return None if sign_bit else 0
    x = pow(x2, (_P + 3) // 8, _P)
    if (x * x - x2) % _P != 0:
        x = x * _SQRT_M1 % _P
    if (x * x - x2) % _P != 0:
        return None
    if (x & 1) != sign_bit:
        x = _P - x
    return x


_G_Y = 4 * _inv(5) % _P
_G_X = _recover_x(_G_Y, 0)
# Extended homogeneous coordinates: (X, Y, Z, T) with x = X/Z, y = Y/Z, xy = T/Z.
_G = (_G_X, _G_Y, 1, _G_X * _G_Y % _P)
_IDENTITY = (0, 1, 1, 0)


def _point_add(p, q):
    a = (p[1] - p[0]) * (q[1] - q[0]) % _P
    b = (p[1] + p[0]) * (q[1] + q[0]) % _P
    c = 2 * p[3] * q[3] * _D % _P
    d = 2 * p[2] * q[2] % _P
    e, f, g, h = b - a, d - c, d + c, b + a
    return (e * f % _P, g * h % _P, f * g % _P, e * h % _P)


def _point_mul(scalar, point):
    out = _IDENTITY
    while scalar > 0:
        if scalar & 1:
            out = _point_add(out, point)
        point = _point_add(point, point)
        scalar >>= 1
    return out


def _point_equal(p, q):
    if (p[0] * q[2] - q[0] * p[2]) % _P != 0:
        return False
    return (p[1] * q[2] - q[1] * p[2]) % _P == 0


def _point_compress(p):
    z_inv = _inv(p[2])
    x = p[0] * z_inv % _P
    y = p[1] * z_inv % _P
    return (y | ((x & 1) << 255)).to_bytes(32, "little")


def _point_decompress(raw):
    if len(raw) != 32:
        return None
    y = int.from_bytes(raw, "little")
    sign_bit = y >> 255
    y &= (1 << 255) - 1
    x = _recover_x(y, sign_bit)
    if x is None:
        return None
    return (x, y, 1, x * y % _P)


def _secret_expand(seed):
    if len(seed) != 32:
        raise ValueError("Ed25519 seed must be exactly 32 bytes")
    digest = _sha512(seed)
    scalar = int.from_bytes(digest[:32], "little")
    scalar &= (1 << 254) - 8
    scalar |= 1 << 254
    return scalar, digest[32:]


def ed25519_public_key(seed):
    """32-byte seed -> 32-byte public key."""
    scalar, _ = _secret_expand(seed)
    return _point_compress(_point_mul(scalar, _G))


def ed25519_sign(seed, message):
    """32-byte seed + message bytes -> 64-byte signature."""
    scalar, prefix = _secret_expand(seed)
    pub = _point_compress(_point_mul(scalar, _G))
    r = int.from_bytes(_sha512(prefix + message), "little") % _L
    big_r = _point_compress(_point_mul(r, _G))
    k = int.from_bytes(_sha512(big_r + pub + message), "little") % _L
    s = (r + k * scalar) % _L
    return big_r + s.to_bytes(32, "little")


def ed25519_verify(public_key, message, signature):
    """True only for a well-formed signature that checks out. Never raises."""
    try:
        if len(public_key) != 32 or len(signature) != 64:
            return False
        point_a = _point_decompress(public_key)
        if point_a is None:
            return False
        r_raw = signature[:32]
        point_r = _point_decompress(r_raw)
        if point_r is None:
            return False
        s = int.from_bytes(signature[32:], "little")
        if s >= _L:
            return False
        k = int.from_bytes(_sha512(r_raw + public_key + message), "little") % _L
        return _point_equal(_point_mul(s, _G),
                            _point_add(point_r, _point_mul(k, point_a)))
    except (ValueError, TypeError):
        return False
# --- END ED25519 SHARED CORE ---

# ---------------------------------------------------------------------------
# License verification
# ---------------------------------------------------------------------------

# The production signing key for ciscotools.dev license grants. Replace the
# empty string with the hex public key printed by
# `python -m backend.services.license_grants --new-key` at release time; the
# app refuses every grant it cannot trace to a trusted key.
PROD_PUBLIC_KEY_HEX = "7d498657a7dc1206d1942ea537251836151d1ddf005c2684b0ff72d31492820f"

# !!! DEVELOPMENT KEY - NOT A SECRET !!!
# The matching seed is published in the server source so the suite and a
# local run work with no setup. A grant signed with it is REFUSED unless the
# operator sets ALLOW_DEV_LICENSE_KEY=true in config.env.
DEV_PUBLIC_KEY_HEX = "ff57575dc7af8bfc4d0837cc1ce2017b686a88145dc5579a958e3462fe9a908e"

# Environment-only escape hatch for developing against a dev license server.
# Deliberately NOT a config.env field: see LicenseClient.dev_override_key.
DEV_OVERRIDE_ENV = "NETOPS_DEV_LICENSE_KEY"

UNPAID_BANNER = "DEV KEY, NOT A PAID LICENSE."

GRACE_DAYS = 30
# Tolerance for a machine whose clock is a little behind the server's.
CLOCK_SKEW_SECONDS = 24 * 3600

LICENSE_PATH = "/api/v1/license"


def _utcnow():
    return datetime.datetime.now(datetime.timezone.utc)


def parse_iso8601(text):
    """Parse the server's timestamps. Returns None for anything unexpected.

    datetime.fromisoformat on 3.9 does not accept a trailing Z, and the app
    has to run on stock macOS python3, so the Z is normalised by hand.
    """
    if not isinstance(text, str) or not text:
        return None
    cleaned = text.strip()
    if cleaned.endswith("Z"):
        cleaned = cleaned[:-1] + "+00:00"
    try:
        moment = datetime.datetime.fromisoformat(cleaned)
    except ValueError:
        return None
    if moment.tzinfo is None:
        moment = moment.replace(tzinfo=datetime.timezone.utc)
    return moment.astimezone(datetime.timezone.utc)


def _b64url_decode(text):
    pad = "=" * (-len(text) % 4)
    import base64

    return base64.urlsafe_b64decode(text + pad)


def verify_grant_token(token, trusted_keys):
    """(grant dict, key_hex) for a token signed by one of `trusted_keys`.

    Returns (None, reason) for anything else. Never raises: a truncated
    token, a tampered payload, and a signature from an untrusted key all end
    the same way, with the Pro cards locked.
    """
    if not isinstance(token, str) or "." not in token:
        return None, "malformed grant"
    payload_b64, _, signature_b64 = token.partition(".")
    if not payload_b64 or not signature_b64:
        return None, "malformed grant"
    try:
        payload = _b64url_decode(payload_b64)
        signature = _b64url_decode(signature_b64)
    except Exception:
        return None, "malformed grant"
    for key_hex in trusted_keys:
        if not key_hex:
            continue
        try:
            key = bytes.fromhex(key_hex)
        except ValueError:
            continue
        if ed25519_verify(key, payload, signature):
            try:
                grant = json.loads(payload.decode("utf-8"))
            except Exception:
                return None, "malformed grant"
            if not isinstance(grant, dict):
                return None, "malformed grant"
            return grant, key_hex
    return None, "signature does not match a trusted license key"


class LicenseState(object):
    """What the UI needs to know: is Pro unlocked, and if not, why not."""

    def __init__(self, pro=False, reason="No license key configured.",
                 grant=None, needs_refresh=True, key_hex=""):
        self.pro = pro
        self.reason = reason
        self.grant = grant or {}
        self.needs_refresh = needs_refresh
        self.key_hex = key_hex

    def as_dict(self):
        return {
            "pro": self.pro,
            "reason": self.reason,
            "account": self.grant.get("account", ""),
            "plan": self.grant.get("plan", ""),
            "issued": self.grant.get("issued", ""),
            "expires": self.grant.get("expires", ""),
            "key_id": self.grant.get("key_id", ""),
            "needs_refresh": self.needs_refresh,
            "dev_key": bool(self.key_hex) and self.key_hex == DEV_PUBLIC_KEY_HEX,
            # True whenever Pro is unlocked by anything other than the
            # production key. The UI shouts about it; see UNPAID_BANNER.
            "unpaid": bool(self.pro) and not (
                PROD_PUBLIC_KEY_HEX and self.key_hex == PROD_PUBLIC_KEY_HEX),
        }


def evaluate_grant(grant, key_hex, now=None):
    """Decide whether a verified grant still unlocks Pro, right now.

    Three separate clocks have to line up and any one of them can lock the
    app: the grant's own expiry, the 30-day offline grace from `issued`, and
    a sanity check that the machine's clock is not set before the grant was
    issued (which is how you would try to live in the grace window forever).
    """
    now = now or _utcnow()
    issued = parse_iso8601(grant.get("issued"))
    expires = parse_iso8601(grant.get("expires"))
    if issued is None or expires is None:
        return False, "grant is missing a usable issued/expires timestamp", True
    if grant.get("plan") != "pro":
        return False, "this account is not on the Pro plan", True
    if now < issued - datetime.timedelta(seconds=CLOCK_SKEW_SECONDS):
        return False, ("this machine's clock is set before the license was "
                       "issued; fix the clock and refresh"), True
    grace_end = issued + datetime.timedelta(days=GRACE_DAYS)
    if now >= grace_end:
        return False, "offline grace period has run out; refresh the license", True
    if now >= expires:
        return False, "license grant has expired; refresh the license", True
    remaining = min(expires, grace_end) - now
    days = max(0, int(remaining.total_seconds() // 86400))
    return True, "Pro unlocked (%d day%s left before the next check)" % (
        days, "" if days == 1 else "s"), False


class LicenseClient(object):
    """Fetches, caches, and verifies the offline grant.

    `opener` is injected so the test suite can exercise every path (valid,
    invalid key, lapsed subscription, revoked key, malformed body, server
    unreachable) without a socket ever being opened.
    """

    def __init__(self, config, opener=None, now=None):
        self.config = config
        self.cache_path = config.path("LICENSE_CACHE")
        self.api_key = config.get("CISCOTOOLS_API_KEY", "")
        self.base_url = config.get("CISCOTOOLS_BASE_URL",
                                   "https://ciscotools.dev").rstrip("/")
        self.allow_dev = config.bool("ALLOW_DEV_LICENSE_KEY", False)
        self._opener = opener or self._http_get
        self._now = now or _utcnow

    # -- trusted keys -------------------------------------------------------

    def dev_override_key(self):
        """A build-your-own trust anchor, for developing against a dev server.

        Environment ONLY, never config.env: a setting an engineer can fill in
        from a documented example file is not a developer tool, it is a
        paywall with a form field. It is also refused outright once a
        production key is compiled in, so a shipped build cannot be talked
        into trusting a self-minted grant without editing the source (which
        anyone who owns the file can do anyway, and which leaves a trace).
        """
        raw = (os.environ.get(DEV_OVERRIDE_ENV) or "").strip()
        if not raw or PROD_PUBLIC_KEY_HEX:
            return ""
        if not re.match(r"^[0-9a-fA-F]{64}$", raw):
            return ""
        return raw.lower()

    def trusted_keys(self):
        """Keys whose signature unlocks Pro, in order of authority."""
        keys = []
        if PROD_PUBLIC_KEY_HEX:
            keys.append(PROD_PUBLIC_KEY_HEX)
        if self.allow_dev:
            keys.append(DEV_PUBLIC_KEY_HEX)
        override = self.dev_override_key()
        if override:
            keys.append(override)
        return keys

    def is_paid_key(self, key_hex):
        return bool(PROD_PUBLIC_KEY_HEX) and key_hex == PROD_PUBLIC_KEY_HEX

    # -- cache --------------------------------------------------------------

    def read_cache(self):
        try:
            with open(self.cache_path, "r", encoding="utf-8") as handle:
                blob = json.load(handle)
        except (OSError, ValueError):
            return ""
        token = blob.get("grant_token") if isinstance(blob, dict) else None
        return token if isinstance(token, str) else ""

    def write_cache(self, token, extra=None):
        blob = {"grant_token": token, "cached_at": _iso(self._now()),
                "app": APP_NAME, "app_version": APP_VERSION}
        if isinstance(extra, dict):
            for name in ("key_id", "grace_days"):
                if name in extra:
                    blob[name] = extra[name]
        directory = os.path.dirname(os.path.abspath(self.cache_path))
        if directory and not os.path.isdir(directory):
            os.makedirs(directory, exist_ok=True)
        temp = self.cache_path + ".tmp"
        with open(temp, "w", encoding="utf-8") as handle:
            json.dump(blob, handle, indent=2)
        os.replace(temp, self.cache_path)
        try:
            os.chmod(self.cache_path, 0o600)
        except OSError:
            pass

    # -- state --------------------------------------------------------------

    def state(self):
        token = self.read_cache()
        if not token:
            if not self.api_key:
                return LicenseState(
                    False,
                    "No API key configured. Put your ciscotools.dev key in "
                    "config.env as CISCOTOOLS_API_KEY to unlock Pro.",
                )
            return LicenseState(False, "No cached license yet. Refresh to fetch one.")
        grant, key_hex = verify_grant_token(token, self.trusted_keys())
        if grant is None:
            return LicenseState(False, key_hex)
        if grant.get("app") not in (APP_NAME, "any"):
            return LicenseState(False, "this grant was issued for another app",
                                grant, True, key_hex)
        ok, reason, needs_refresh = evaluate_grant(grant, key_hex, self._now())
        if ok and not self.is_paid_key(key_hex):
            reason = "%s %s" % (UNPAID_BANNER, reason)
        return LicenseState(ok, reason, grant, needs_refresh, key_hex)

    # -- refresh ------------------------------------------------------------

    def _http_get(self, url, headers, timeout):  # pragma: no cover - network
        request = urllib.request.Request(url, headers=headers, method="GET")
        context = ssl.create_default_context()
        with urllib.request.urlopen(request, timeout=timeout,
                                    context=context) as response:
            return response.status, response.read().decode("utf-8", "replace")

    def refresh(self):
        """Ask the server for a fresh grant. Returns (ok, message)."""
        if not self.api_key:
            return False, ("No API key configured. Add CISCOTOOLS_API_KEY to "
                           "config.env (find it on your account page).")
        query = urllib.parse.urlencode({"app": APP_NAME, "version": APP_VERSION})
        url = "%s%s?%s" % (self.base_url, LICENSE_PATH, query)
        headers = {
            "X-API-Key": self.api_key,
            "Accept": "application/json",
            "User-Agent": "%s/%s" % (APP_NAME, APP_VERSION),
        }
        try:
            status, body = self._opener(url, headers, 20)
        except urllib.error.HTTPError as error:  # pragma: no cover - network
            status = error.code
            try:
                body = error.read().decode("utf-8", "replace")
            except Exception:
                body = ""
        except Exception as error:  # pragma: no cover - network
            return False, "Could not reach %s (%s)." % (self.base_url, error)
        if status == 401:
            return False, ("The server rejected this API key. Issue a new one "
                           "on your account page.")
        if status == 402:
            return False, ("This account is not on the Pro plan right now. "
                           "Lite keeps working.")
        if status != 200:
            return False, "License server returned HTTP %s." % status
        try:
            payload = json.loads(body)
        except ValueError:
            return False, "License server returned a response this app could not read."
        token = payload.get("grant_token") if isinstance(payload, dict) else None
        if not isinstance(token, str) or not token:
            return False, "License server returned no grant."
        grant, key_hex = verify_grant_token(token, self.trusted_keys())
        if grant is None:
            hint = ""
            if key_hex.startswith("signature") and not PROD_PUBLIC_KEY_HEX:
                hint = (" This build has no production license key compiled in; "
                        "set ALLOW_DEV_LICENSE_KEY=true only against a dev server.")
            return False, "Refused the grant: %s.%s" % (key_hex, hint)
        # Scope check BEFORE the cache write. state() has always enforced this;
        # refresh() used to cache the grant and report "Pro unlocked" while
        # every Pro card stayed locked, which is a lie the UI then repeats.
        if grant.get("app") not in (APP_NAME, "any"):
            return False, ("Refused the grant: it was issued for another app "
                           "(%r), not this one." % str(grant.get("app"))[:40])
        ok, reason, _needs = evaluate_grant(grant, key_hex, self._now())
        if not ok:
            return False, "Refused the grant: %s." % reason
        self.write_cache(token, payload)
        if not self.is_paid_key(key_hex):
            reason = "%s %s" % (UNPAID_BANNER, reason)
        return True, reason


# ---------------------------------------------------------------------------
# config.env
# ---------------------------------------------------------------------------

CONFIG_DEFAULTS = {
    "SSH_USERNAME": "",
    "SSH_PASSWORD": "",
    "ENABLE_PASSWORD": "",
    "SSH_PORT": "22",
    "SSH_TIMEOUT": "30",
    "SSH_STRICT_HOST_KEY_CHECKING": "accept-new",
    "INVENTORY": "inventory.csv",
    "OUTPUT_DIR": "./output",
    "JOBS_DIR": "./jobs",
    "AUDIT_LOG": "./netops-audit.log",
    "DRY_RUN": "true",
    "CISCOTOOLS_API_KEY": "",
    "CISCOTOOLS_BASE_URL": "https://ciscotools.dev",
    "LICENSE_CACHE": "./license.json",
    "ALLOW_DEV_LICENSE_KEY": "false",
    "BIND_HOST": "127.0.0.1",
    "BIND_PORT": "8781",
    "OPEN_BROWSER": "true",
}

# Names whose values must never reach a log line, an audit record, or the UI.
SECRET_KEYS = ("SSH_PASSWORD", "ENABLE_PASSWORD", "CISCOTOOLS_API_KEY")

TRUE_WORDS = ("1", "true", "yes", "on")
FALSE_WORDS = ("0", "false", "no", "off", "")


def parse_config_env(text):
    """Parse a config.env body into a dict. Tolerant, never raises.

    Accepts KEY=value, ignores blank lines and # comments, strips one layer
    of matching quotes, and keeps a trailing inline comment only when the
    value is unquoted (so a password containing a # is not truncated).
    """
    values = {}
    for raw_line in text.splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if line.startswith("export "):
            line = line[len("export "):].strip()
        if "=" not in line:
            continue
        name, _, value = line.partition("=")
        name = name.strip()
        if not name or not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", name):
            continue
        value = value.strip()
        if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
            value = value[1:-1]
        values[name] = value
    return values


class Config(object):
    def __init__(self, values=None, base_dir=None):
        self.base_dir = os.path.abspath(base_dir or os.getcwd())
        self.values = dict(CONFIG_DEFAULTS)
        if values:
            self.values.update(values)
        self.unknown = sorted(set(values or {}) - set(CONFIG_DEFAULTS))

    @classmethod
    def load(cls, path):
        base_dir = os.path.dirname(os.path.abspath(path)) or os.getcwd()
        try:
            with open(path, "r", encoding="utf-8") as handle:
                values = parse_config_env(handle.read())
        except OSError:
            values = {}
        # Environment wins over the file, so a CI run or a wrapper script can
        # override without editing config.env.
        for name in CONFIG_DEFAULTS:
            if os.environ.get(name):
                values[name] = os.environ[name]
        return cls(values, base_dir)

    def get(self, name, default=""):
        value = self.values.get(name, default)
        return value if value is not None else default

    def bool(self, name, default=False):
        raw = str(self.get(name, "")).strip().lower()
        if raw in TRUE_WORDS:
            return True
        if raw in FALSE_WORDS:
            return False if raw != "" else default
        return default

    def int(self, name, default):
        try:
            return int(str(self.get(name, "")).strip())
        except (TypeError, ValueError):
            return default

    def path(self, name):
        raw = self.get(name, "")
        if not raw:
            raw = CONFIG_DEFAULTS.get(name, "")
        if os.path.isabs(raw):
            return raw
        return os.path.normpath(os.path.join(self.base_dir, raw))

    def redacted(self):
        out = {}
        for name, value in sorted(self.values.items()):
            if name in SECRET_KEYS:
                out[name] = "(set)" if value else "(not set)"
            else:
                out[name] = value
        return out

    def prompt_for_missing_secrets(self, interactive=True, prompter=None):
        """Blank password fields are asked for at startup, never stored."""
        prompter = prompter or getpass.getpass
        if not interactive:
            return
        if not self.get("SSH_USERNAME"):
            return
        if not self.get("SSH_PASSWORD"):
            try:
                self.values["SSH_PASSWORD"] = prompter(
                    "SSH password for %s (leave blank for key auth): "
                    % self.get("SSH_USERNAME"))
            except (EOFError, KeyboardInterrupt):
                self.values["SSH_PASSWORD"] = ""


# ---------------------------------------------------------------------------
# Inventory
# ---------------------------------------------------------------------------

HOSTNAME_RE = re.compile(r"^[A-Za-z0-9]([A-Za-z0-9._-]{0,61}[A-Za-z0-9])?$")
GROUP_RE = re.compile(r"^[A-Za-z0-9._-]{0,32}$")
PLATFORMS = ("ios", "iosxe", "nxos", "iosxr", "asa", "other")

RESERVED_COLUMNS = ("hostname", "ip", "platform", "group")


class Device(object):
    def __init__(self, hostname, ip, platform="ios", group="", variables=None):
        self.hostname = hostname
        self.ip = ip
        self.platform = platform
        self.group = group
        self.variables = variables or {}

    def as_dict(self):
        out = {"hostname": self.hostname, "ip": self.ip,
               "platform": self.platform, "group": self.group}
        out.update({"var_" + k: v for k, v in sorted(self.variables.items())})
        return out

    def context(self):
        """Variables available to a config template for this device."""
        context = dict(self.variables)
        context.update({"hostname": self.hostname, "ip": self.ip,
                        "platform": self.platform, "group": self.group})
        return context


# Byte-order marks Excel writes. "Unicode Text (*.txt)" is UTF-16 with a BOM
# and is a perfectly ordinary thing for someone to hand this app.
_BOM_UTF8 = b"\xef\xbb\xbf"
_BOM_UTF16_LE = b"\xff\xfe"
_BOM_UTF16_BE = b"\xfe\xff"

_ENCODING_HELP = ("Save it from Excel as \"CSV UTF-8 (Comma delimited)\" and "
                  "try again.")


def decode_inventory_bytes(raw):
    """Bytes from an inventory file -> (text, error). Never raises.

    Excel hands people UTF-8-with-BOM and UTF-16 far more often than plain
    UTF-8, and the app used to die on both with a raw UnicodeDecodeError
    before serve() was ever reached. A file this app cannot read is a message,
    not a traceback.
    """
    if not isinstance(raw, (bytes, bytearray)):
        return str(raw), ""
    raw = bytes(raw)
    for bom, encoding in ((_BOM_UTF8, "utf-8-sig"),
                          (_BOM_UTF16_LE, "utf-16"),
                          (_BOM_UTF16_BE, "utf-16")):
        if raw.startswith(bom):
            try:
                return raw.decode(encoding), ""
            except (UnicodeDecodeError, ValueError):
                return "", ("The inventory starts with a %s byte-order mark "
                            "but does not decode as that encoding. %s"
                            % (encoding, _ENCODING_HELP))
    # A BOM-less UTF-16 file is mostly NUL bytes -- and NUL is valid UTF-8,
    # so this heuristic has to run BEFORE the utf-8 attempt or the file
    # "decodes" into text full of NULs that csv then chokes on.
    sample = raw[:512]
    if sample.count(b"\x00") > len(sample) // 4:
        for encoding in ("utf-16-le", "utf-16-be"):
            try:
                decoded = raw.decode(encoding)
            except (UnicodeDecodeError, ValueError):
                continue
            if "\x00" not in decoded:
                return decoded, ""
    try:
        return raw.decode("utf-8"), ""
    except UnicodeDecodeError:
        return "", ("The inventory is not UTF-8 text. %s" % _ENCODING_HELP)


def parse_inventory(text):
    """Parse an inventory CSV. Returns (devices, errors). Never raises.

    Every field is validated rather than trusted: an inventory is a file a
    human edits, and a hostname carrying a newline or a semicolon would end
    up in a .ini file, a filename, and an ssh argument list.

    Accepts bytes as well as text, so a caller can hand over whatever it read
    off disk and let decode_inventory_bytes deal with Excel's encodings.
    """
    devices = []
    errors = []
    if isinstance(text, (bytes, bytearray)):
        text, error = decode_inventory_bytes(text)
        if error:
            return [], [error]
    if not isinstance(text, str):
        return [], ["The inventory could not be read as text."]
    if "\x00" in text:
        # csv.reader raises _csv.Error("line contains NUL") on these, which
        # used to escape all the way out of Engine.__init__.
        line = text[:text.index("\x00")].count("\n") + 1
        return [], ["The inventory contains a NUL byte on line %d, so it is "
                    "not a plain CSV file. %s" % (line, _ENCODING_HELP)]
    try:
        reader = csv.DictReader(io.StringIO(text))
        fieldnames = reader.fieldnames or []
    except csv.Error as error:
        return [], ["Could not read the CSV: %s" % error]
    normalised = {}
    for name in fieldnames:
        if name is None:
            continue
        normalised[name] = name.strip().lower().replace(" ", "_")
    lowered = set(normalised.values())
    for required in ("hostname", "ip"):
        if required not in lowered:
            errors.append("Missing required column: %s" % required)
    if errors:
        return [], errors

    seen = {}
    try:
        rows = list(enumerate(reader, start=2))
    except csv.Error as error:
        return [], ["Could not read the CSV: %s" % error]
    for index, row in rows:
        record = {}
        for raw_name, value in row.items():
            if raw_name is None:
                continue
            record[normalised.get(raw_name, raw_name)] = (value or "").strip()
        hostname = record.get("hostname", "")
        ip_text = record.get("ip", "")
        if not hostname and not ip_text:
            continue  # blank line
        if not HOSTNAME_RE.match(hostname):
            errors.append("Line %d: bad hostname %r (letters, digits, dot, "
                          "dash, underscore only)" % (index, hostname))
            continue
        try:
            ip = str(ipaddress.ip_address(ip_text))
        except ValueError:
            errors.append("Line %d: %r is not a valid IP address" % (index, ip_text))
            continue
        platform = (record.get("platform") or "ios").lower()
        if platform not in PLATFORMS:
            errors.append("Line %d: unknown platform %r (expected one of %s)"
                          % (index, platform, ", ".join(PLATFORMS)))
            continue
        group = record.get("group", "")
        if not GROUP_RE.match(group):
            errors.append("Line %d: bad group %r" % (index, group))
            continue
        key = (hostname.lower(), ip)
        if key in seen:
            errors.append("Line %d: duplicate of line %d (%s / %s)"
                          % (index, seen[key], hostname, ip))
            continue
        seen[key] = index
        variables = {name: value for name, value in record.items()
                     if name not in RESERVED_COLUMNS and name}
        devices.append(Device(hostname, ip, platform, group, variables))
    return devices, errors


def filter_devices(devices, group="", hostnames=None):
    selected = devices
    if group:
        selected = [d for d in selected if d.group == group]
    if hostnames:
        wanted = {h.strip().lower() for h in hostnames if h.strip()}
        selected = [d for d in selected if d.hostname.lower() in wanted]
    return selected


# ---------------------------------------------------------------------------
# Command validation
# ---------------------------------------------------------------------------

# A command reaches a device over an interactive SSH session. Anything that
# could end the line early, start a second command, or smuggle control
# characters into the terminal stream is refused rather than escaped.
_BAD_COMMAND_CHARS = set("\r\n\x00\x1b")

# The only two run modes. `mode` arrives from a POST body and decides three
# separate things (the Pro gate, the confirmation step, and whether an audit
# line is written), so an unrecognised value is refused rather than treated
# as "not config".
MODES = ("show", "config")

# Leading verbs a SHOW run may send. This is an allowlist, not a filter: a
# command has to be recognisably a read before it goes out under a label that
# skips the Pro gate, the confirmation, and the audit line. Anything else
# belongs in config mode, where all three apply.
SHOW_MODE_VERBS = (
    "show", "sh", "display", "dir", "more", "ping", "traceroute", "tracert",
    "trace",
    # Session-local on IOS, NX-OS and ASA; needed to stop the pager.
    "terminal",
)

# `terminal ...` is only session-local for these.
TERMINAL_SUBCOMMANDS = (
    "length", "width", "pager", "monitor", "no", "exec", "dont-ask",
    "session-timeout",
)

# Refused in show mode by name, purely so the engineer is told what is
# actually wrong ("that is a configuration command") instead of the generic
# allowlist message. The allowlist above already refuses every one of them.
CONFIG_MODE_VERBS = (
    "configure", "conf", "config", "write", "wr", "copy", "reload", "erase",
    "delete", "format", "boot", "username", "enable", "no", "default",
    "hostname", "interface", "router", "line", "vlan", "ip", "ipv6", "aaa",
    "snmp-server", "tacacs-server", "radius-server", "crypto", "archive",
    "license", "install", "request", "commit", "rollback", "factory-reset",
    "upgrade", "squeeze", "clear", "debug", "undebug", "test", "tclsh",
    "guestshell", "bash", "end", "exit", "reset", "rmdir", "mkdir", "rename",
)


def validate_command(command, keep_indent=False, mode="config"):
    """Returns (ok, cleaned_or_reason).

    The control-character check runs on the RAW string, before any stripping,
    so a trailing carriage return is a rejection and not something quietly
    tidied away. `keep_indent` preserves leading whitespace, which matters in
    config mode: the plan, the audit line, and the device should all show the
    lines the way the engineer wrote them in the template.

    In SHOW mode the command must also be a read. A run labelled "show" skips
    the Pro gate, the confirmation prompt and the audit line, so the label has
    to be true: without this check a Lite user could type `configure terminal`
    into the extra-commands box, untick dry run, and push an unaudited config.
    """
    if not isinstance(command, str):
        return False, "command must be text"
    for char in command:
        if char in _BAD_COMMAND_CHARS or (ord(char) < 32 and char != "\t"):
            return False, "command contains a control character"
    cleaned = command.rstrip() if keep_indent else command.strip()
    if not cleaned.strip():
        return False, "empty command"
    if len(cleaned) > 512:
        return False, "command longer than 512 characters"
    if mode == "show":
        ok, reason = _check_show_mode(cleaned)
        if not ok:
            return False, reason
    return True, cleaned


# `;` is a COMMAND SEPARATOR on NX-OS ("configure terminal ; interface Eth1/1
# ; no shut" is ordinary syntax there), and `nxos` is one of the platforms the
# inventory accepts. Round 2 judged a command by its first token only, so
# "show version ; configure terminal ; username backdoor ... password X"
# sailed through the read-verb allowlist and pushed config with none of the
# three rails. In a show run `;` is now refused outright, exactly the way CR
# and LF already are: a read has no reason to chain.
#
# Config mode keeps `;` -- it is legitimate NX-OS syntax and that path is
# already Pro, confirmed and audited. The audit line redacts each `;` segment
# on its own so a chained secret cannot hide behind the first one.
#
# `>` and `>>` are output redirection on NX-OS: "show running-config >
# bootflash:cfg.txt" is a read that WRITES a file on the device, under the
# one label that skips the Pro gate, the confirmation and the audit line.
# Same door as "| redirect", one character wide, so it is shut the same way.
#
# The lookalikes are belt and braces. None of them is exploitable today -- the
# app writes raw UTF-8 to the pty, so a device receives the multi-byte
# sequence and no Cisco CLI treats it as a separator -- but U+037E is
# canonically equivalent to ";" under NFC and U+FF1B under NFKC, so any future
# layer that normalises before this check would hand a real semicolon
# straight through. Refusing them costs nothing; a read command has no use
# for any of these characters.
_SHOW_MODE_FORBIDDEN_CHARS = (
    ";",       # ASCII semicolon: NX-OS command separator
    ">",       # output redirection (covers ">>" too)
    "\uff1b",  # FULLWIDTH SEMICOLON
    "\u204f",  # REVERSED SEMICOLON
    "\u037e",  # GREEK QUESTION MARK, NFC-equivalent to ";"
)

# Percent-encoded semicolon. Nothing in this app URL-decodes a command, so
# this is literal text today; refused for the same reason as the lookalikes.
_SHOW_MODE_FORBIDDEN_TEXT = ("%3b",)

# Pipe stages that WRITE. "show running-config | redirect bootflash:x" is a
# read command that creates a file on the device, which is not what a show run
# is allowed to mean. Filters (include, exclude, section, begin, count, json)
# are untouched.
_WRITING_PIPE_STAGES = ("redirect", "tee", "append", "file")


def _check_show_mode(cleaned):
    for char in _SHOW_MODE_FORBIDDEN_CHARS:
        if char in cleaned:
            if char == ">":
                return False, (
                    "'>' redirects output to a file on the device (NX-OS "
                    "writes it), so it is not allowed in a show run. Filters "
                    "such as include, exclude, section and begin are fine.")
            return False, (
                "%r chains commands on some platforms (NX-OS treats it as a "
                "command separator), so it is not allowed in a show run. Send "
                "one read per line, or use the config card for changes." % char)
    lowered_all = cleaned.lower()
    for text in _SHOW_MODE_FORBIDDEN_TEXT:
        if text in lowered_all:
            return False, (
                "%r is an encoded command separator and is not allowed in a "
                "show run." % text)
    lowered = cleaned.lower()
    if "|" in lowered:
        for stage in lowered.split("|")[1:]:
            head = stage.strip().split()
            if head and head[0] in _WRITING_PIPE_STAGES:
                return False, (
                    "a show run cannot pipe to %r: that writes a file on the "
                    "device. Filters such as include, exclude, section and "
                    "begin are fine." % head[0])
    tokens = cleaned.strip().lower().split()
    verb = tokens[0]
    if verb in CONFIG_MODE_VERBS:
        return False, (
            "%r is a configuration command and cannot run in a show run. Use "
            "the config card, which is Pro, asks for confirmation, and writes "
            "an audit line." % verb)
    if verb not in SHOW_MODE_VERBS:
        return False, (
            "%r is not a read-only command. A show run only sends: %s."
            % (verb, ", ".join(sorted(SHOW_MODE_VERBS))))
    if verb == "terminal":
        if len(tokens) < 2 or tokens[1] not in TERMINAL_SUBCOMMANDS:
            return False, (
                "only terminal %s are allowed in a show run"
                % "/".join(sorted(TERMINAL_SUBCOMMANDS)))
    if verb in ("ping", "traceroute", "tracert", "trace") and len(tokens) < 2:
        return False, ("%s needs a target; the bare form opens an interactive "
                       "dialog the runner cannot answer" % verb)
    return True, ""


def validate_commands(commands, keep_indent=False, mode="config"):
    cleaned = []
    errors = []
    for command in commands:
        ok, result = validate_command(command, keep_indent=keep_indent,
                                      mode=mode)
        if ok:
            cleaned.append(result)
        else:
            errors.append("%r: %s" % (command, result))
    return cleaned, errors


SHOW_PRESETS = [
    ("show version", "Software, uptime, and hardware"),
    ("show inventory", "Chassis, modules, transceivers"),
    ("show ip interface brief", "Interface addressing and state"),
    ("show interfaces status", "Switchport status, VLAN, duplex, speed"),
    ("show cdp neighbors detail", "Neighbours with addresses and platforms"),
    ("show lldp neighbors detail", "LLDP equivalent for mixed-vendor edges"),
    ("show running-config", "Full running configuration"),
    ("show vlan brief", "VLAN table"),
    ("show ip route summary", "Routing table size by protocol"),
    ("show mac address-table", "Learned MAC addresses"),
    ("show logging last 100", "Recent syslog on the box"),
    ("show environment", "Power, fans, temperature"),
]


# ---------------------------------------------------------------------------
# Config templates (Pro)
# ---------------------------------------------------------------------------

TEMPLATE_VAR_RE = re.compile(r"\{\{\s*([A-Za-z_][A-Za-z0-9_]*)\s*\}\}")


def template_variables(template):
    return sorted(set(TEMPLATE_VAR_RE.findall(template or "")))


def render_template(template, context):
    """Substitute {{VAR}} from `context`. Returns (lines, missing).

    Substitution is a single pass over the template: a value that itself
    contains {{...}} is NOT expanded again, so an inventory column can never
    be used to inject a second round of template syntax.
    """
    missing = []
    lowered = {str(k).lower(): v for k, v in (context or {}).items()}

    def replace(match):
        name = match.group(1)
        if name.lower() in lowered:
            return str(lowered[name.lower()])
        missing.append(name)
        return match.group(0)

    rendered = TEMPLATE_VAR_RE.sub(replace, template or "")
    lines = [line.rstrip() for line in rendered.splitlines()]
    lines = [line for line in lines if line.strip()]
    return lines, sorted(set(missing))


# ---------------------------------------------------------------------------
# Job planning
# ---------------------------------------------------------------------------

def build_plan(devices, commands, mode="show", dry_run=True,
               template="", write_mem=False):
    """The dry-run plan: exactly what would happen, device by device.

    Building the plan never opens a connection, so the plan view is safe to
    refresh as often as you like, including with DRY_RUN off.
    """
    steps = []
    problems = []
    if mode not in MODES:
        # No steps at all, so a caller that ignores `problems` still sends
        # nothing. Engine.run() refuses on problems before it connects.
        return {
            "mode": str(mode)[:32], "dry_run": True, "device_count": 0,
            "command_count": 0, "steps": [],
            "problems": ["Unknown mode %r. Expected one of: %s."
                         % (str(mode)[:32], ", ".join(MODES))],
            "would_change_devices": False,
        }
    for device in devices:
        if mode == "config" and template:
            lines, missing = render_template(template, device.context())
            if missing:
                problems.append("%s: template variable(s) with no value: %s"
                                % (device.hostname, ", ".join(missing)))
            device_commands = lines
        else:
            device_commands = list(commands)
        cleaned, errors = validate_commands(device_commands,
                                            keep_indent=(mode == "config"),
                                            mode=mode)
        for error in errors:
            problems.append("%s: %s" % (device.hostname, error))
        steps.append({
            "hostname": device.hostname,
            "ip": device.ip,
            "platform": device.platform,
            "group": device.group,
            "mode": mode,
            "commands": cleaned,
            "write_mem": bool(write_mem) and mode == "config",
        })
    return {
        "mode": mode,
        "dry_run": bool(dry_run),
        "device_count": len(steps),
        "command_count": sum(len(step["commands"]) for step in steps),
        "steps": steps,
        "problems": problems,
        "would_change_devices": mode == "config" and not dry_run,
    }


# ---------------------------------------------------------------------------
# Transport
# ---------------------------------------------------------------------------

class RunResult(object):
    def __init__(self, hostname, ip, ok, output="", error="", duration=0.0,
                 per_command=None):
        self.hostname = hostname
        self.ip = ip
        self.ok = ok
        self.output = output
        self.error = error
        self.duration = duration
        self.per_command = per_command or {}

    def as_dict(self):
        return {"hostname": self.hostname, "ip": self.ip, "ok": self.ok,
                "error": self.error, "duration": round(self.duration, 2),
                "bytes": len(self.output)}


class Transport(object):
    """Interface the runner talks to. The tests substitute a fake."""

    def run(self, device, commands, config_mode=False, write_mem=False):
        raise NotImplementedError


class FakeTransport(Transport):
    """Scripted transport for tests and for `--self-test`. Opens nothing."""

    def __init__(self, responses=None, fail_hosts=()):
        self.responses = responses or {}
        self.fail_hosts = set(fail_hosts)
        self.calls = []

    def run(self, device, commands, config_mode=False, write_mem=False):
        self.calls.append({"hostname": device.hostname, "commands": list(commands),
                           "config_mode": config_mode, "write_mem": write_mem})
        if device.hostname in self.fail_hosts:
            return RunResult(device.hostname, device.ip, False,
                             error="scripted failure")
        chunks = []
        per_command = {}
        for command in commands:
            body = self.responses.get((device.hostname, command))
            if body is None:
                body = self.responses.get(command, "%s: no scripted output" % command)
            per_command[command] = body
            chunks.append("%s# %s\n%s" % (device.hostname, command, body))
        return RunResult(device.hostname, device.ip, True,
                         output="\n".join(chunks), per_command=per_command)


# Prompt shapes, matched against the tail of the device stream. A leading
# CR/LF is required so a fragment of ordinary output cannot look like a
# prompt; the tail always carries one by the time a prompt is printed.
_PASSWORD_PROMPT_RE = re.compile(rb"(?i)(password|passcode)\s*:\s*$")
_PRIV_PROMPT_RE = re.compile(rb"[\r\n][-\w.@():/]+#\s*$")
_EXEC_PROMPT_RE = re.compile(rb"[\r\n][-\w.@():/]+>\s*$")
_PROMPT_RE = re.compile(rb"[\r\n][-\w.@():/]+[>#]\s*$")


def classify_prompt(tail):
    """What the device is asking for: password / exec / privileged / "".

    Split out of the pty loop so the state machine that drives the login and
    the `enable` handshake is testable without a device or a pty: the loop
    itself can only be exercised against real hardware.

    exec is the `sw01>` prompt (privilege 1, cannot configure), privileged is
    `sw01#` (and `sw01(config)#`, which the same pattern matches because the
    parens are inside the character class).
    """
    if isinstance(tail, str):
        tail = tail.encode("utf-8", "replace")
    if not isinstance(tail, (bytes, bytearray)):
        return ""
    stripped = bytes(tail).rstrip()
    if _PASSWORD_PROMPT_RE.search(stripped):
        return "password"
    if _PRIV_PROMPT_RE.search(stripped):
        return "privileged"
    if _EXEC_PROMPT_RE.search(stripped):
        return "exec"
    return ""


class SSHTransport(Transport):
    """Drives the system `ssh` client over a pty.

    Why a pty and not paramiko: the app must be a single stdlib-only file, so
    there is no SSH library to import. Why a pty and not plain pipes: network
    devices prompt for a password on the terminal, and OpenSSH refuses to read
    one from a pipe. `pty` is in the standard library on macOS and Linux.

    The password is written to the pty and nowhere else. It never appears in
    argv (which is world readable in ps), never in an environment variable,
    and never in a log line.
    """

    PROMPT_RE = _PROMPT_RE
    PASSWORD_RE = _PASSWORD_PROMPT_RE

    def __init__(self, config):
        self.config = config
        self.username = config.get("SSH_USERNAME", "")
        self._password = config.get("SSH_PASSWORD", "")
        self._enable = config.get("ENABLE_PASSWORD", "")
        self.port = config.int("SSH_PORT", 22)
        self.timeout = config.int("SSH_TIMEOUT", 30)
        self.strict = config.get("SSH_STRICT_HOST_KEY_CHECKING", "accept-new")

    def ssh_argv(self, device):
        if self.strict not in ("yes", "no", "accept-new", "ask"):
            strict = "accept-new"
        else:
            strict = self.strict
        argv = [
            "ssh",
            "-p", str(self.port),
            "-o", "StrictHostKeyChecking=%s" % strict,
            "-o", "ConnectTimeout=%d" % max(5, min(self.timeout, 120)),
            "-o", "NumberOfPasswordPrompts=1",
            "-o", "PubkeyAuthentication=yes",
        ]
        if self.username:
            argv += ["-l", self.username]
        argv.append(device.ip)
        return argv

    def run(self, device, commands, config_mode=False, write_mem=False):
        import pty  # imported here so a Windows run can still use Lite/CLI

        started = time.time()
        if shutil.which("ssh") is None:
            return RunResult(device.hostname, device.ip, False,
                             error="no `ssh` client found on PATH")
        script = list(commands)
        if config_mode:
            script = ["configure terminal"] + script + ["end"]
            if write_mem:
                script.append("write memory")
        master, slave = pty.openpty()
        try:
            process = subprocess.Popen(
                self.ssh_argv(device), stdin=slave, stdout=slave, stderr=slave,
                close_fds=True, start_new_session=True,
            )
        except OSError as error:
            os.close(master)
            os.close(slave)
            return RunResult(device.hostname, device.ip, False,
                             error="could not start ssh: %s" % error)
        os.close(slave)
        collected = bytearray()
        sent_password = False
        # Enable handshake state. ENABLE_PASSWORD used to be documented,
        # prompted for, stored, and never sent, so on any device not already
        # at privilege 15 `configure terminal` was refused and the run just
        # timed out with no hint why.
        sent_enable = False
        sent_enable_password = False
        deadline = started + max(20, self.timeout * max(2, len(script)))
        queue = list(script) + ["exit"]
        try:
            os.set_blocking(master, False)
            while time.time() < deadline:
                if process.poll() is not None and not _readable(master, 0.2):
                    break
                if _readable(master, 0.4):
                    try:
                        chunk = os.read(master, 65536)
                    except (OSError, BlockingIOError):
                        chunk = b""
                    if not chunk:
                        break
                    collected += chunk
                tail = bytes(collected[-256:])
                prompt = classify_prompt(tail)
                if prompt == "password" and sent_enable and not sent_enable_password:
                    # The second password prompt of the session belongs to
                    # `enable`, not to the login.
                    os.write(master, self._enable.encode() + b"\n")
                    sent_enable_password = True
                    collected += b"\n"
                    continue
                if prompt == "exec" and self._enable and not sent_enable:
                    os.write(master, b"enable\n")
                    sent_enable = True
                    time.sleep(0.05)
                    continue
                if not sent_password and prompt == "password":
                    if not self._password:
                        return RunResult(
                            device.hostname, device.ip, False,
                            error="device asked for a password and none is set")
                    os.write(master, self._password.encode() + b"\n")
                    sent_password = True
                    collected += b"\n"
                    continue
                if queue and prompt in ("exec", "privileged"):
                    command = queue.pop(0)
                    os.write(master, command.encode() + b"\n")
                    time.sleep(0.05)
            ok = not queue
            error = "" if ok else "timed out before every command was sent"
        finally:
            try:
                process.terminate()
            except OSError:
                pass
            try:
                os.close(master)
            except OSError:
                pass
        text = self._scrub_credentials(collected.decode("utf-8", "replace"))
        return RunResult(device.hostname, device.ip, ok, output=text,
                         error=error, duration=time.time() - started)

    def _scrub_credentials(self, text):
        """Remove the credentials THIS transport just sent from the stream.

        The password goes to the pty, and a pty can echo. OpenSSH turns echo
        off while it reads a password, so on a real login this is usually a
        no-op -- but "usually" is not a property to hand a transcript file,
        and the app knows the exact strings, so there is no guesswork. Runs
        before the RunResult exists, so every consumer (the output file, the
        UI, the before/after diff) sees the scrubbed copy.
        """
        for secret in (self._password, self._enable):
            if secret and len(secret) >= 3:
                text = text.replace(secret, "***")
        return text


def _readable(fd, timeout):
    import select

    try:
        ready, _, _ = select.select([fd], [], [], timeout)
    except (OSError, ValueError):
        return False
    return bool(ready)


# ---------------------------------------------------------------------------
# Audit log
# ---------------------------------------------------------------------------

_AUDIT_LOCK = threading.Lock()


def _iso(moment=None):
    moment = moment or _utcnow()
    return moment.replace(microsecond=0).isoformat().replace("+00:00", "Z")


# A device command carries its secret as a positional argument, not under a
# field name, so redact() cannot see it by key. The list is deliberately a
# little over-eager: losing the argument of "crypto key generate" from an
# audit line costs nothing, and keeping a TACACS key in a file on someone's
# laptop costs a lot.
#
# The trailing lookahead accepts `:` `=` and `,` as well as whitespace, so
# "Secret: Tr0ub4dor" in banner prose is caught the same as "secret X".
_SECRET_ARG_RE = re.compile(
    r"(?i)(?:^|[\s;])(password|passwd|passphrase|secret|community|"
    r"key-string|key|pre-shared-key|shared-secret|psk|wpa-psk|md5|"
    r"authentication-key|auth-key|encryption|credential)(?=[\s:=,]|$)")

_REDACTED = "***"

# Fields whose value is a device command (or a list of them).
_COMMAND_FIELDS = ("commands", "command", "config_lines", "lines")

# Multi-line blocks whose BODY is free prose. A banner can carry a secret with
# no keyword anywhere near it ("the root pw is Tr0ub4dor"), so once one opens
# every line up to its delimiter is dropped wholesale rather than scanned.
# Group 1 is the header through the delimiter, 2 the delimiter, 3 the rest.
_BANNER_OPEN_RE = re.compile(r"(?i)^(\s*banner\s+\S+\s+(\S+))(.*)$")

# `certificate self-signed 01` ... `quit` is the other block shape that
# carries opaque material line by line.
_QUIT_BLOCK_OPEN_RE = re.compile(r"(?i)^\s*certificate(\s|$)")


# Punctuation that can sit between a keyword and its value, or trail a bare
# prompt. Stripping it decides whether anything actually FOLLOWS the keyword.
_SECRET_PUNCTUATION = " \t:=,"


def _segment_shape(segment):
    """(match, has_value, has_prefix) for one command segment.

    Three shapes, and the difference decides how much is kept:

      "snmp-server community S3cret RO"  the keyword names a parameter and the
                                         VALUE follows -> drop the tail.
      "Hunter2 is the password"          the keyword is last, so whatever it
                                         refers to came BEFORE -> drop it all.
      "Password: "                       the keyword is the whole segment: a
                                         device's own prompt, with no secret
                                         on it at all -> leave it alone.
    """
    if not isinstance(segment, str):
        return None, False, False
    match = _SECRET_ARG_RE.search(segment)
    if match is None:
        return None, False, False
    has_value = bool(segment[match.end(1):].strip(_SECRET_PUNCTUATION).strip())
    has_prefix = bool(segment[:match.start(1)].strip(_SECRET_PUNCTUATION).strip())
    return match, has_value, has_prefix


def _segment_secret(segment):
    """The part of one command segment that must not be kept. "" if none."""
    match, has_value, has_prefix = _segment_shape(segment)
    if match is None:
        return ""
    if has_value:
        return segment[match.end(1):].strip()
    if has_prefix:
        return segment.strip()
    return ""


def _redact_segment(segment):
    match, has_value, has_prefix = _segment_shape(segment)
    if match is None:
        return segment
    if has_value:
        return segment[:match.end(1)] + " " + _REDACTED
    if has_prefix:
        return _REDACTED
    # Nothing before it and nothing after it: a bare "Password:" prompt echoed
    # back by the device. Rewriting that to "Password ***" only made a
    # transcript read oddly; there is no secret on the line to remove.
    return segment


def redact_command(text):
    """Scrub one device command.

    `;` is a command separator on NX-OS, so each segment is judged on its own:
    a secret chained behind a harmless first segment cannot hide behind it.
    """
    if not isinstance(text, str):
        return text
    if ";" not in text:
        return _redact_segment(text)
    return "".join(part if part == ";" else _redact_segment(part)
                   for part in re.split(r"(;)", text))


def _banner_block(line):
    """(header, delimiter, remainder) when `line` opens a banner block."""
    match = _BANNER_OPEN_RE.match(line)
    if match is None:
        return None
    header, delimiter, remainder = match.group(1), match.group(2), match.group(3)
    if delimiter in remainder:
        return None  # single-line banner: the body is right here, not a block
    return header, delimiter, remainder


def redact_command_list(commands):
    """Redact a list of commands, treating multi-line blocks as blocks."""
    out = []
    block = None
    for item in commands:
        if not isinstance(item, str):
            out.append(item)
            continue
        if block is not None:
            kind, delimiter = block
            if kind == "quit" and item.strip().lower() == "quit":
                out.append(item)
                block = None
            elif kind == "banner" and delimiter in item:
                out.append(_REDACTED)
                block = None
            else:
                out.append(_REDACTED)
            continue
        opened = _banner_block(item)
        if opened is not None:
            header, delimiter, remainder = opened
            out.append(header + (" " + _REDACTED if remainder.strip() else ""))
            block = ("banner", delimiter)
            continue
        if _QUIT_BLOCK_OPEN_RE.match(item):
            out.append(item)
            block = ("quit", "")
            continue
        out.append(redact_command(item))
    return out


def secret_fragments(commands):
    """Every substring the command redaction dropped, for scrubbing elsewhere.

    The per-device transcript is the device echoing back what we sent, so it
    holds the same secrets. Rather than guess at the structure of a terminal
    stream, the exact values we know we sent are removed from it by name.
    """
    fragments = set()
    block = None
    for item in commands:
        if not isinstance(item, str):
            continue
        if block is not None:
            kind, delimiter = block
            if kind == "quit" and item.strip().lower() == "quit":
                block = None
            else:
                body = item.strip()
                # A bare delimiter or a two-character marker is not a secret,
                # and replacing it across a transcript is pure noise.
                if len(body) >= 3 and body != delimiter:
                    fragments.add(body)
                if kind == "banner" and delimiter in item:
                    block = None
            continue
        opened = _banner_block(item)
        if opened is not None:
            _header, delimiter, remainder = opened
            if len(remainder.strip()) >= 3:
                fragments.add(remainder.strip())
            block = ("banner", delimiter)
            continue
        if _QUIT_BLOCK_OPEN_RE.match(item):
            block = ("quit", "")
            continue
        for part in re.split(r"(;)", item):
            if part == ";":
                continue
            secret = _segment_secret(part)
            if len(secret) >= 3:
                fragments.add(secret)
    return fragments


def redact_transcript(text, commands):
    """Scrub a CONFIG-PUSH transcript of the secrets we sent.

    Applied only to config-mode runs. A show run's transcript is the device's
    own output and is the product -- scrubbing a `show running-config` dump
    would break the feature people download this for -- so it is written as
    the device printed it. The README says exactly that.

    Two passes: the exact values we know we sent are removed wherever they
    appear (including the device's echo), then every line is run through the
    command scrubber to catch anything the device echoed differently.
    """
    if not isinstance(text, str) or not text:
        return text
    out = text
    for fragment in sorted(secret_fragments(commands), key=len, reverse=True):
        out = out.replace(fragment, _REDACTED)
    lines = []
    for line in out.split("\n"):
        suffix = "\r" if line.endswith("\r") else ""
        body = line[:-1] if suffix else line
        lines.append(redact_command(body) + suffix)
    return "\n".join(lines)


def redact(record):
    """Strip anything that smells like a credential, at any depth."""
    if isinstance(record, dict):
        out = {}
        for name, value in record.items():
            lowered = str(name).lower()
            if any(word in lowered for word in
                   ("password", "passwd", "secret", "api_key", "apikey",
                    "token", "credential")):
                out[name] = "***"
            elif lowered in _COMMAND_FIELDS:
                if isinstance(value, (list, tuple)):
                    out[name] = redact_command_list(list(value))
                else:
                    out[name] = redact_command(value)
            else:
                out[name] = redact(value)
        return out
    if isinstance(record, list):
        return [redact(item) for item in record]
    return record


def append_audit(path, record):
    """Append ONE json line. Called only after a real write succeeded."""
    entry = redact(dict(record))
    entry.setdefault("at", _iso())
    entry.setdefault("app", APP_NAME)
    entry.setdefault("app_version", APP_VERSION)
    line = json.dumps(entry, sort_keys=True) + "\n"
    directory = os.path.dirname(os.path.abspath(path))
    if directory and not os.path.isdir(directory):
        os.makedirs(directory, exist_ok=True)
    with _AUDIT_LOCK:
        # 0600 at creation: this file records what was pushed to devices, and
        # even after redaction that is nobody else's business on a shared
        # machine. An existing file keeps whatever mode its owner chose.
        handle = _open_private(path, "a")
        try:
            handle.write(line)
        finally:
            handle.close()
    return entry


def _open_private(path, mode="w"):
    """open() that creates the file 0600 instead of 0644."""
    flags = os.O_WRONLY | os.O_CREAT
    flags |= os.O_APPEND if "a" in mode else os.O_TRUNC
    fd = os.open(path, flags, 0o600)
    return os.fdopen(fd, mode, encoding="utf-8")


# ---------------------------------------------------------------------------
# Output files
# ---------------------------------------------------------------------------

_UNSAFE_NAME_RE = re.compile(r"[^A-Za-z0-9._-]")


def safe_filename(name, fallback="device"):
    """One path COMPONENT, with no way out of the directory it lands in."""
    cleaned = _UNSAFE_NAME_RE.sub("_", str(name or "").strip())
    cleaned = cleaned.lstrip(".")
    cleaned = cleaned[:64]
    return cleaned or fallback


def safe_archive_name(*parts):
    """Build a zip entry path that cannot escape the extraction directory.

    This is the Zip Slip guard: every component is reduced to a safe
    filename, so "..", "/etc/passwd", "C:\\x", and a name carrying a
    backslash all collapse to something harmless before they are written.
    """
    cleaned = [safe_filename(part) for part in parts if str(part or "").strip()]
    cleaned = [part for part in cleaned if part not in ("", ".", "..")]
    if not cleaned:
        cleaned = ["item"]
    return "/".join(cleaned)


def write_device_output(output_dir, run_id, result, sent_commands=None):
    """Write one device's transcript. 0600, and scrubbed for a config push.

    `sent_commands` is passed for a CONFIG run only. The transcript is the
    device echoing back what we sent, so without this it holds the community
    string and the `username ... password X` line in cleartext while the audit
    line next to it says `***` (critic finding R4).

    A SHOW run's transcript is left exactly as the device printed it: it is
    the device's own output and it is the whole point of the feature. The
    README says so in as many words.
    """
    directory = os.path.join(output_dir, safe_filename(run_id, "run"))
    os.makedirs(directory, exist_ok=True)
    path = os.path.join(directory, safe_filename(result.hostname) + ".txt")
    body = result.output
    if sent_commands:
        body = redact_transcript(body, list(sent_commands))
    # 0600: this is a running-config dump on someone's laptop.
    handle = _open_private(path, "w")
    try:
        handle.write(body)
    finally:
        handle.close()
    return path


# ---------------------------------------------------------------------------
# SecureCRT session export
# ---------------------------------------------------------------------------

_INI_STRIP_RE = re.compile(r"[\x00-\x1f\x7f]")


def ini_escape(value):
    """SecureCRT .ini values are one line. Control characters are removed.

    Without this a hostname of "sw01\\nS:\"Username\"=root" would append a
    second directive to the session file. The inventory parser already
    refuses such a hostname; this is the second lock on the same door.
    """
    return _INI_STRIP_RE.sub("", str(value or ""))


def ini_dword(number):
    return "%08x" % (int(number) & 0xFFFFFFFF)


def build_session_ini(device, username="", port=22, protocol="SSH2",
                      emulation="Xterm"):
    lines = [
        "S:\"Hostname\"=%s" % ini_escape(device.ip),
        "S:\"Protocol Name\"=%s" % ini_escape(protocol),
        "D:\"[SSH2] Port\"=%s" % ini_dword(port),
        "D:\"Port\"=%s" % ini_dword(port),
        "S:\"Emulation\"=%s" % ini_escape(emulation),
        "S:\"Description\"=%s" % ini_escape(
            "%s (%s) exported by %s %s"
            % (device.hostname, device.platform, APP_TITLE, APP_VERSION)),
        "D:\"Session Password Saved\"=%s" % ini_dword(0),
        "D:\"Auth Prompts in Window\"=%s" % ini_dword(1),
    ]
    if username:
        lines.append("S:\"Username\"=%s" % ini_escape(username))
    # A trailing newline keeps SecureCRT from complaining about the last key.
    return "\r\n".join(lines) + "\r\n"


def build_sessions_csv(devices):
    buffer = io.StringIO()
    writer = csv.writer(buffer, lineterminator="\n")
    writer.writerow(["hostname", "ip", "platform", "group", "session_path"])
    for device in devices:
        writer.writerow([
            device.hostname, device.ip, device.platform, device.group,
            safe_archive_name("Sessions", device.group or "CiscoTools",
                              device.hostname + ".ini"),
        ])
    return buffer.getvalue()


def build_session_zip(devices, username="", port=22):
    """Zip of one .ini per device plus a sessions.csv index. Bytes, in memory."""
    buffer = io.BytesIO()
    with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
        seen = set()
        for device in devices:
            name = safe_archive_name("Sessions", device.group or "CiscoTools",
                                     device.hostname + ".ini")
            candidate = name
            counter = 2
            while candidate in seen:
                candidate = name[:-4] + ("_%d.ini" % counter)
                counter += 1
            seen.add(candidate)
            archive.writestr(candidate, build_session_ini(device, username, port))
        archive.writestr("sessions.csv", build_sessions_csv(devices))
        archive.writestr("README.txt", SESSION_README)
    return buffer.getvalue()


SESSION_README = (
    "SecureCRT sessions exported by %s %s\r\n"
    "\r\n"
    "Import on macOS:\r\n"
    "  1. Quit SecureCRT.\r\n"
    "  2. In Finder press Shift-Cmd-G and go to\r\n"
    "     ~/Library/Application Support/VanDyke/SecureCRT/Config/Sessions/\r\n"
    "  3. Copy the Sessions folder from this zip into that folder.\r\n"
    "  4. Start SecureCRT. The sessions appear in the Session Manager.\r\n"
    "\r\n"
    "Import on Windows:\r\n"
    "  Copy the Sessions folder into\r\n"
    "  %%APPDATA%%\\VanDyke\\Config\\Sessions\\\r\n"
    "\r\n"
    "No password is stored in these files. SecureCRT will prompt you.\r\n"
) % (APP_TITLE, APP_VERSION)


# ---------------------------------------------------------------------------
# Saved job bundles (Pro)
# ---------------------------------------------------------------------------

JOB_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._-]{0,48}$")


class JobStore(object):
    def __init__(self, directory):
        self.directory = directory

    def _path(self, name):
        return os.path.join(self.directory, safe_filename(name, "job") + ".json")

    def save(self, name, spec):
        if not JOB_NAME_RE.match(str(name or "")):
            return None, ("Job names are letters, digits, space, dot, dash, "
                          "underscore, up to 49 characters.")
        os.makedirs(self.directory, exist_ok=True)
        blob = dict(spec)
        blob["name"] = name
        blob["saved_at"] = _iso()
        blob["app_version"] = APP_VERSION
        blob.pop("password", None)
        path = self._path(name)
        handle = _open_private(path, "w")
        try:
            json.dump(redact(blob), handle, indent=2, sort_keys=True)
        finally:
            handle.close()
        return path, None

    def load(self, name):
        try:
            with open(self._path(name), "r", encoding="utf-8") as handle:
                blob = json.load(handle)
        except (OSError, ValueError):
            return None
        return blob if isinstance(blob, dict) else None

    def list(self):
        try:
            names = sorted(os.listdir(self.directory))
        except OSError:
            return []
        out = []
        for entry in names:
            if not entry.endswith(".json"):
                continue
            blob = self.load(entry[:-5])
            if blob:
                out.append({"name": blob.get("name", entry[:-5]),
                            "saved_at": blob.get("saved_at", ""),
                            "mode": blob.get("mode", "show"),
                            "device_count": len(blob.get("hostnames", []) or [])})
        return out

    def delete(self, name):
        try:
            os.remove(self._path(name))
            return True
        except OSError:
            return False


# ---------------------------------------------------------------------------
# Scheduling (Pro)
# ---------------------------------------------------------------------------

_TIME_OF_DAY_RE = re.compile(r"^([01]?\d|2[0-3]):([0-5]\d)$")


def next_run_at(schedule, now=None):
    """Pure function so the schedule maths is testable without waiting.

    schedule is {"kind": "interval", "minutes": N} or
    {"kind": "daily", "at": "HH:MM"}. Returns a datetime or None.
    """
    now = now or _utcnow()
    if not isinstance(schedule, dict):
        return None
    kind = schedule.get("kind")
    if kind == "interval":
        try:
            minutes = int(schedule.get("minutes", 0))
        except (TypeError, ValueError):
            return None
        if minutes < 1 or minutes > 7 * 24 * 60:
            return None
        return now + datetime.timedelta(minutes=minutes)
    if kind == "daily":
        match = _TIME_OF_DAY_RE.match(str(schedule.get("at", "")))
        if not match:
            return None
        hour, minute = int(match.group(1)), int(match.group(2))
        candidate = now.replace(hour=hour, minute=minute, second=0, microsecond=0)
        if candidate <= now:
            candidate += datetime.timedelta(days=1)
        return candidate
    return None


class Scheduler(object):
    """Runs saved jobs on a timer, in this process, while the app is open.

    Deliberately not a cron writer: the app never installs anything on the
    machine, so closing it stops every scheduled run, which is the behaviour
    an engineer expects from a tool they double-clicked.
    """

    def __init__(self, runner, clock=None, sleeper=None):
        self.runner = runner
        self.entries = {}
        self.history = []
        self._clock = clock or _utcnow
        self._sleep = sleeper or time.sleep
        self._lock = threading.Lock()
        self._stop = threading.Event()
        self._thread = None

    def add(self, name, job, schedule):
        when = next_run_at(schedule, self._clock())
        if when is None:
            return None, "Schedule must be {kind: interval, minutes: N} or {kind: daily, at: HH:MM}."
        with self._lock:
            self.entries[name] = {"job": job, "schedule": schedule, "next": when}
        return when, None

    def remove(self, name):
        with self._lock:
            return self.entries.pop(name, None) is not None

    def listing(self):
        with self._lock:
            return [{"name": name, "next": _iso(entry["next"]),
                     "schedule": entry["schedule"]}
                    for name, entry in sorted(self.entries.items())]

    def tick(self):
        """Run anything due. Returns the names it ran. Called by the thread
        and directly by the tests."""
        now = self._clock()
        due = []
        with self._lock:
            for name, entry in self.entries.items():
                if entry["next"] <= now:
                    due.append(name)
        for name in due:
            with self._lock:
                entry = self.entries.get(name)
            if not entry:
                continue
            try:
                outcome = self.runner(entry["job"])
            except Exception as error:  # noqa: BLE001 - a bad job must not kill the timer
                outcome = {"ok": False, "error": str(error)}
            with self._lock:
                entry["next"] = next_run_at(entry["schedule"], self._clock())
                if entry["next"] is None:
                    self.entries.pop(name, None)
            self.history.append({"name": name, "at": _iso(now), "outcome": outcome})
            self.history = self.history[-50:]
        return due

    def start(self):  # pragma: no cover - thread wrapper
        if self._thread:
            return
        def loop():
            while not self._stop.is_set():
                try:
                    self.tick()
                except Exception:
                    pass
                self._stop.wait(20)
        self._thread = threading.Thread(target=loop, daemon=True)
        self._thread.start()

    def stop(self):  # pragma: no cover - thread wrapper
        self._stop.set()


# ---------------------------------------------------------------------------
# The engine the UI and the CLI both call
# ---------------------------------------------------------------------------

class Engine(object):
    def __init__(self, config, transport=None, license_client=None):
        self.config = config
        self.transport = transport
        self.license = license_client or LicenseClient(config)
        self.jobs = JobStore(config.path("JOBS_DIR"))
        self.devices = []
        self.inventory_errors = []
        self.last_run = None
        self.before = {}
        self.scheduler = Scheduler(self.run_saved_job)
        self.load_inventory()

    # -- inventory ----------------------------------------------------------

    def load_inventory(self, text=None):
        """Load and parse the inventory. NEVER raises: a bad file has to
        leave the app running with an error on the page, because this runs
        from Engine.__init__ and a raise here means the app does not start."""
        if text is None:
            path = self.config.path("INVENTORY")
            try:
                with open(path, "rb") as handle:
                    text = handle.read()
            except OSError:
                self.devices, self.inventory_errors = [], []
                return self.devices, self.inventory_errors
        try:
            self.devices, self.inventory_errors = parse_inventory(text)
        except Exception as error:  # noqa: BLE001 - startup must survive
            self.devices = []
            self.inventory_errors = ["Could not read the inventory: %s" % error]
        return self.devices, self.inventory_errors

    def groups(self):
        return sorted({device.group for device in self.devices if device.group})

    # -- gating -------------------------------------------------------------

    def pro(self):
        return self.license.state().pro

    def _require_pro(self):
        state = self.license.state()
        if state.pro:
            return None
        return {"ok": False, "pro_required": True,
                "error": "This is a Pro feature. %s" % state.reason}

    # -- transport ----------------------------------------------------------

    def _transport(self):
        if self.transport is None:
            self.transport = SSHTransport(self.config)
        return self.transport

    # -- the runs -----------------------------------------------------------

    def plan(self, selection, commands, mode="show", template="",
             write_mem=False, dry_run=None):
        devices = filter_devices(self.devices, selection.get("group", ""),
                                 selection.get("hostnames"))
        if dry_run is None:
            dry_run = self.config.bool("DRY_RUN", True)
        return build_plan(devices, commands, mode=mode, dry_run=dry_run,
                          template=template, write_mem=write_mem)

    def run(self, selection, commands, mode="show", template="",
            write_mem=False, dry_run=None, confirm=False, label=""):
        """Execute a plan. Returns a JSON-friendly result envelope.

        A dry run returns the plan and stops. A config-mode run additionally
        requires `confirm`, so a stray POST cannot change a device.

        `mode` is validated against MODES first. It decides the Pro gate, the
        confirmation and the audit line, so an unrecognised value must be an
        error, never a quiet "not config".
        """
        if mode not in MODES:
            return {"ok": False,
                    "error": "Unknown mode %r. Expected one of: %s."
                             % (str(mode)[:32], ", ".join(MODES))}
        if mode == "config":
            gate = self._require_pro()
            if gate:
                return gate
        plan = self.plan(selection, commands, mode=mode, template=template,
                         write_mem=write_mem, dry_run=dry_run)
        if plan["problems"]:
            return {"ok": False, "error": "Fix the plan first.", "plan": plan}
        if not plan["steps"]:
            return {"ok": False, "error": "No devices selected.", "plan": plan}
        if plan["dry_run"]:
            return {"ok": True, "dry_run": True, "plan": plan,
                    "message": "Dry run: nothing was sent. %d device(s), %d command(s)."
                               % (plan["device_count"], plan["command_count"])}
        if mode == "config" and not confirm:
            return {"ok": False, "confirm_required": True, "plan": plan,
                    "error": "This would change %d device(s). Confirm to proceed."
                             % plan["device_count"]}

        transport = self._transport()
        run_id = "%s-%s" % (datetime.datetime.now().strftime("%Y%m%d-%H%M%S"),
                            safe_filename(label or mode, mode))
        output_dir = self.config.path("OUTPUT_DIR")
        results = []
        by_host = {device.hostname: device for device in self.devices}
        for step in plan["steps"]:
            device = by_host.get(step["hostname"])
            if device is None:
                continue
            result = transport.run(device, step["commands"],
                                   config_mode=(mode == "config"),
                                   write_mem=step["write_mem"])
            try:
                path = write_device_output(
                    output_dir, run_id, result,
                    sent_commands=(step["commands"] if mode == "config"
                                   else None))
            except OSError as error:
                path = ""
                result.error = result.error or ("could not write output: %s" % error)
            record = result.as_dict()
            record["output_file"] = path
            results.append(record)
            # The audit line is written AFTER the write succeeded, never
            # before, and only for runs that actually changed something.
            if mode == "config" and result.ok:
                append_audit(self.config.path("AUDIT_LOG"), {
                    "action": "config_push",
                    "run_id": run_id,
                    "hostname": device.hostname,
                    "ip": device.ip,
                    "platform": device.platform,
                    "commands": step["commands"],
                    "write_mem": step["write_mem"],
                    "account": self.license.state().grant.get("account", ""),
                })
        self.last_run = {"run_id": run_id, "mode": mode, "results": results}
        ok_count = sum(1 for record in results if record["ok"])
        return {"ok": True, "dry_run": False, "run_id": run_id,
                "results": results, "output_dir": output_dir,
                "message": "%d of %d device(s) completed."
                           % (ok_count, len(results))}

    # -- before/after diff (Pro) --------------------------------------------

    def capture_before(self, selection, commands):
        gate = self._require_pro()
        if gate:
            return gate
        outcome = self.run(selection, commands, mode="show", dry_run=False,
                           label="before")
        if not outcome.get("ok"):
            return outcome
        self.before = {}
        for record in outcome["results"]:
            path = record.get("output_file")
            if path:
                try:
                    with open(path, "r", encoding="utf-8") as handle:
                        self.before[record["hostname"]] = handle.read()
                except OSError:
                    pass
        return {"ok": True, "captured": sorted(self.before),
                "message": "Captured 'before' output from %d device(s)."
                           % len(self.before)}

    def diff_after(self, selection, commands):
        gate = self._require_pro()
        if gate:
            return gate
        if not self.before:
            return {"ok": False, "error": "Capture a 'before' snapshot first."}
        outcome = self.run(selection, commands, mode="show", dry_run=False,
                           label="after")
        if not outcome.get("ok"):
            return outcome
        diffs = []
        for record in outcome["results"]:
            hostname = record["hostname"]
            before = self.before.get(hostname)
            if before is None:
                continue
            after = ""
            path = record.get("output_file")
            if path:
                try:
                    with open(path, "r", encoding="utf-8") as handle:
                        after = handle.read()
                except OSError:
                    after = ""
            lines = list(difflib.unified_diff(
                before.splitlines(), after.splitlines(),
                fromfile="%s before" % hostname, tofile="%s after" % hostname,
                lineterm="", n=2))
            diffs.append({"hostname": hostname, "changed": bool(lines),
                          "diff": "\n".join(lines)})
        changed = sum(1 for entry in diffs if entry["changed"])
        return {"ok": True, "diffs": diffs,
                "message": "%d of %d device(s) changed." % (changed, len(diffs))}

    # -- saved jobs (Pro) ---------------------------------------------------

    def save_job(self, name, spec):
        gate = self._require_pro()
        if gate:
            return gate
        path, error = self.jobs.save(name, spec)
        if error:
            return {"ok": False, "error": error}
        return {"ok": True, "path": path, "message": "Saved job %r." % name}

    def run_saved_job(self, name_or_spec):
        spec = name_or_spec
        if isinstance(name_or_spec, str):
            spec = self.jobs.load(name_or_spec)
            if spec is None:
                return {"ok": False, "error": "No saved job by that name."}
        selection = {"group": spec.get("group", ""),
                     "hostnames": spec.get("hostnames") or None}
        return self.run(
            selection, spec.get("commands") or [],
            mode=spec.get("mode", "show"), template=spec.get("template", ""),
            write_mem=bool(spec.get("write_mem")),
            dry_run=spec.get("dry_run", True), confirm=bool(spec.get("confirm")),
            label=spec.get("name", "job"))

    # -- SecureCRT ----------------------------------------------------------

    def session_zip(self, selection):
        devices = filter_devices(self.devices, selection.get("group", ""),
                                 selection.get("hostnames"))
        return build_session_zip(devices, self.config.get("SSH_USERNAME", ""),
                                 self.config.int("SSH_PORT", 22)), len(devices)

    # -- status -------------------------------------------------------------

    def status(self):
        state = self.license.state()
        return {
            "app": APP_NAME,
            "title": APP_TITLE,
            "version": APP_VERSION,
            "python": "%d.%d.%d" % sys.version_info[:3],
            "license": state.as_dict(),
            "dry_run_default": self.config.bool("DRY_RUN", True),
            "inventory_path": self.config.path("INVENTORY"),
            "device_count": len(self.devices),
            "inventory_errors": self.inventory_errors,
            "groups": self.groups(),
            "output_dir": self.config.path("OUTPUT_DIR"),
            "audit_log": self.config.path("AUDIT_LOG"),
            "presets": [{"command": c, "help": h} for c, h in SHOW_PRESETS],
            "devices": [device.as_dict() for device in self.devices],
            "jobs": self.jobs.list() if state.pro else [],
            "schedules": self.scheduler.listing() if state.pro else [],
            "config": self.config.redacted(),
        }


# ---------------------------------------------------------------------------
# Local web UI
# ---------------------------------------------------------------------------

PAGE_HTML = """<!DOCTYPE html>
<html lang="en"><head><meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>__TITLE__</title>
<style>
:root{--ink:#10151c;--muted:#5b6673;--line:#dde3ea;--bg:#f5f7fa;--card:#fff;
--accent:#0ea5e9;--accent-dark:#0284c7;--warn:#b45309;--ok:#15803d;--lock:#94a3b8}
*{box-sizing:border-box}
body{margin:0;font:15px/1.55 system-ui,-apple-system,Segoe UI,Roboto,sans-serif;
background:var(--bg);color:var(--ink)}
header{background:var(--ink);color:#fff;padding:14px 20px;display:flex;
align-items:center;gap:14px;flex-wrap:wrap}
header h1{font-size:1.05rem;margin:0;font-weight:650}
header .ver{color:#93a4b8;font-size:.8rem}
header .spacer{flex:1}
.pill{font-size:.72rem;padding:3px 9px;border-radius:999px;border:1px solid #2b3648}
.pill.pro{background:#064e3b;border-color:#065f46;color:#6ee7b7}
.pill.lite{background:#1f2937;border-color:#374151;color:#cbd5e1}
.pill.unpaid{background:#7c2d12;border-color:#9a3412;color:#fed7aa}
.unpaidbar{background:#7c2d12;color:#ffedd5;padding:9px 20px;font-size:.82rem;
font-weight:600;letter-spacing:.02em}
main{max-width:1060px;margin:0 auto;padding:20px}
.card{background:var(--card);border:1px solid var(--line);border-radius:10px;
padding:16px 18px;margin-bottom:16px}
.card h2{margin:0 0 4px;font-size:1rem}
.card p.hint{margin:0 0 12px;color:var(--muted);font-size:.85rem}
.card.locked{opacity:.72}
.lockbar{background:#f1f5f9;border:1px dashed var(--lock);color:var(--muted);
padding:8px 10px;border-radius:7px;font-size:.82rem;margin-bottom:12px}
label{display:block;font-size:.8rem;color:var(--muted);margin:10px 0 3px}
input,select,textarea{width:100%;padding:8px 10px;border:1px solid var(--line);
border-radius:7px;font:inherit;background:#fff;color:var(--ink)}
textarea{font-family:ui-monospace,SFMono-Regular,Menlo,monospace;font-size:.82rem;min-height:110px}
button{padding:8px 14px;border:0;border-radius:7px;background:var(--accent);
color:#fff;font:inherit;font-weight:600;cursor:pointer}
button.secondary{background:#e2e8f0;color:var(--ink)}
button.danger{background:var(--warn)}
button:disabled{background:#cbd5e1;cursor:not-allowed}
.row{display:flex;gap:10px;flex-wrap:wrap;align-items:flex-end}
.row>*{flex:1 1 180px}
.actions{display:flex;gap:8px;flex-wrap:wrap;margin-top:12px}
.actions>*{flex:0 0 auto}
pre{background:#0f172a;color:#e2e8f0;padding:12px;border-radius:8px;
overflow:auto;font-size:.78rem;max-height:340px;white-space:pre-wrap;word-break:break-word}
table{width:100%;border-collapse:collapse;font-size:.83rem}
th,td{text-align:left;padding:6px 8px;border-bottom:1px solid var(--line)}
th{color:var(--muted);font-weight:600}
.msg{margin-top:10px;padding:9px 11px;border-radius:7px;font-size:.85rem}
.msg.ok{background:#ecfdf5;color:var(--ok)}
.msg.err{background:#fef2f2;color:#b91c1c}
.msg.warn{background:#fffbeb;color:var(--warn)}
.checks{display:flex;flex-wrap:wrap;gap:6px 14px;margin-top:6px}
.checks label{display:flex;gap:6px;align-items:center;color:var(--ink);
font-size:.82rem;margin:0}
.checks input{width:auto}
footer{color:var(--muted);font-size:.8rem;text-align:center;padding:10px 20px 34px}
footer a{color:var(--accent-dark)}
@media (max-width:640px){main{padding:12px}.card{padding:13px}}
</style></head><body>
<header>
  <h1>__TITLE__</h1><span class="ver">v__VERSION__</span>
  <span class="spacer"></span>
  <span id="tier" class="pill lite">Lite</span>
</header>
<div id="unpaid" class="unpaidbar" hidden></div>
<main>
  <div class="card">
    <h2>Status</h2>
    <p class="hint">Everything below runs on this machine. Nothing is sent to ciscotools.dev
    except the license check, which carries your API key and this app's version and nothing else.</p>
    <div id="status"></div>
    <div class="actions">
      <button class="secondary" onclick="refreshStatus()">Reload inventory</button>
      <button class="secondary" onclick="refreshLicense()">Refresh license</button>
    </div>
    <div id="statusmsg"></div>
  </div>

  <div class="card">
    <h2>Inventory</h2>
    <p class="hint">From <code id="invpath"></code> - columns hostname, ip, platform, group.
    Any extra column becomes a per-device template variable.</p>
    <div id="inventory"></div>
  </div>

  <div class="card">
    <h2>Show-command run <span style="color:var(--muted);font-weight:400;font-size:.8rem">Lite</span></h2>
    <p class="hint">Output is written per device under the output folder. Dry run sends nothing.</p>
    <div class="row">
      <div><label for="group">Group</label><select id="group"></select></div>
      <div><label for="hosts">Only these hostnames (comma separated, blank = all)</label>
        <input id="hosts" placeholder="sw01, sw02"></div>
    </div>
    <label>Preset commands</label>
    <div class="checks" id="presets"></div>
    <label for="extra">Extra show commands, one per line (read-only verbs only)</label>
    <textarea id="extra" placeholder="show ip arp"></textarea>
    <div class="checks"><label><input type="checkbox" id="dryrun" checked> Dry run (send nothing)</label></div>
    <div class="actions">
      <button onclick="doPlan()">Show plan</button>
      <button onclick="doRun()">Run</button>
      <button class="secondary" onclick="exportSessions()">Export SecureCRT sessions</button>
    </div>
    <div id="runmsg"></div>
    <pre id="runout" hidden></pre>
  </div>

  <div class="card" id="configcard">
    <h2>Config push from a template <span style="color:var(--muted);font-weight:400;font-size:.8rem">Pro</span></h2>
    <p class="hint">{{VAR}} placeholders are filled from the device row. A real push asks for
    confirmation and writes one line to the audit log per device, after it succeeds.</p>
    <div id="configlock"></div>
    <label for="template">Config template</label>
    <textarea id="template" placeholder="interface {{UPLINK}}&#10; description {{hostname}} uplink"></textarea>
    <div class="checks">
      <label><input type="checkbox" id="cfgdry" checked> Dry run</label>
      <label><input type="checkbox" id="writemem"> write memory after</label>
    </div>
    <div class="actions">
      <button onclick="doConfigPlan()">Show plan</button>
      <button class="danger" onclick="doConfigPush()">Push config</button>
      <button class="secondary" onclick="captureBefore()">Capture before</button>
      <button class="secondary" onclick="diffAfter()">Diff after</button>
    </div>
    <div id="cfgmsg"></div>
    <pre id="cfgout" hidden></pre>
  </div>

  <div class="card" id="jobcard">
    <h2>Saved jobs and schedules <span style="color:var(--muted);font-weight:400;font-size:.8rem">Pro</span></h2>
    <p class="hint">A saved job is this page's current selection. A schedule runs it while this app stays open.</p>
    <div id="joblock"></div>
    <div class="row">
      <div><label for="jobname">Job name</label><input id="jobname" placeholder="nightly-inventory"></div>
      <div><label for="sched">Schedule</label><input id="sched" placeholder="every 60m  or  daily 02:30"></div>
    </div>
    <div class="actions">
      <button onclick="saveJob()">Save job</button>
      <button class="secondary" onclick="scheduleJob()">Schedule it</button>
    </div>
    <div id="jobmsg"></div>
    <div id="joblist"></div>
  </div>
</main>
<footer>
  __TITLE__ v__VERSION__ &middot; runs entirely on your machine &middot;
  <a href="https://ciscotools.dev/configurator" target="_blank" rel="noopener">docs</a>
</footer>
<script>
const TOKEN = "__TOKEN__";
let STATUS = null;

function esc(s){return String(s==null?'':s).replace(/[&<>"']/g,c=>({'&':'&amp;','<':'&lt;','>':'&gt;','"':'&quot;',"'":'&#39;'}[c]));}
function msg(id, text, kind){document.getElementById(id).innerHTML =
  text ? '<div class="msg '+(kind||'ok')+'">'+esc(text)+'</div>' : '';}
function show(id, text){const el=document.getElementById(id);
  el.hidden = !text; el.textContent = text||'';}

async function api(path, body){
  const opts = {headers:{'X-NetOps-Token':TOKEN}};
  if(body){opts.method='POST';opts.headers['Content-Type']='application/json';
    opts.body=JSON.stringify(body);}
  const res = await fetch(path, opts);
  try { return await res.json(); }
  catch (e) { return {ok:false, error:'HTTP '+res.status}; }
}

function selection(){
  const hosts = document.getElementById('hosts').value.split(',').map(s=>s.trim()).filter(Boolean);
  return {group: document.getElementById('group').value, hostnames: hosts.length?hosts:null};
}
function chosenCommands(){
  const out = [];
  document.querySelectorAll('#presets input:checked').forEach(el=>out.push(el.value));
  document.getElementById('extra').value.split('\\n').forEach(line=>{
    if(line.trim()) out.push(line.trim());});
  return out;
}

function renderStatus(s){
  STATUS = s;
  const lic = s.license;
  const tier = document.getElementById('tier');
  tier.textContent = lic.pro ? (lic.unpaid ? 'Pro (DEV)' : 'Pro') : 'Lite';
  tier.className = 'pill ' + (lic.pro ? (lic.unpaid ? 'unpaid' : 'pro') : 'lite');
  renderUnpaid(lic);
  document.getElementById('invpath').textContent = s.inventory_path;
  document.getElementById('status').innerHTML =
    '<table><tr><th>Python</th><td>'+esc(s.python)+'</td></tr>'+
    '<tr><th>Devices</th><td>'+s.device_count+'</td></tr>'+
    '<tr><th>Dry run default</th><td>'+(s.dry_run_default?'on':'off')+'</td></tr>'+
    '<tr><th>Output folder</th><td>'+esc(s.output_dir)+'</td></tr>'+
    '<tr><th>Audit log</th><td>'+esc(s.audit_log)+'</td></tr>'+
    '<tr><th>License</th><td>'+esc(lic.reason)+(lic.account?' ('+esc(lic.account)+')':'')+'</td></tr></table>';
  const g = document.getElementById('group');
  g.innerHTML = '<option value="">All groups</option>' +
    s.groups.map(x=>'<option>'+esc(x)+'</option>').join('');
  let rows = s.devices.map(d=>'<tr><td>'+esc(d.hostname)+'</td><td>'+esc(d.ip)+
    '</td><td>'+esc(d.platform)+'</td><td>'+esc(d.group)+'</td></tr>').join('');
  document.getElementById('inventory').innerHTML = s.device_count
    ? '<table><tr><th>Hostname</th><th>IP</th><th>Platform</th><th>Group</th></tr>'+rows+'</table>'
    : '<div class="msg warn">No devices loaded. Point INVENTORY in config.env at a CSV.</div>';
  if(s.inventory_errors.length){
    document.getElementById('inventory').innerHTML +=
      '<div class="msg err">'+s.inventory_errors.map(esc).join('<br>')+'</div>';}
  if(!document.getElementById('presets').children.length){
    document.getElementById('presets').innerHTML = s.presets.map(p=>
      '<label><input type="checkbox" value="'+esc(p.command)+'"> '+esc(p.command)+'</label>').join('');
  }
  for(const [cardId, lockId] of [['configcard','configlock'],['jobcard','joblock']]){
    const card = document.getElementById(cardId);
    card.classList.toggle('locked', !lic.pro);
    document.getElementById(lockId).innerHTML = lic.pro ? '' :
      '<div class="lockbar">Locked. '+esc(lic.reason)+
      ' Add CISCOTOOLS_API_KEY to config.env and press Refresh license.</div>';
  }
  document.getElementById('joblist').innerHTML = (s.jobs||[]).length
    ? '<table><tr><th>Job</th><th>Mode</th><th>Saved</th></tr>'+
      s.jobs.map(j=>'<tr><td>'+esc(j.name)+'</td><td>'+esc(j.mode)+'</td><td>'+esc(j.saved_at)+'</td></tr>').join('')+
      '</table>' : '';
}

function renderUnpaid(lic){
  const bar = document.getElementById('unpaid');
  bar.hidden = !lic.unpaid;
  bar.textContent = lic.unpaid
    ? 'DEV KEY, NOT A PAID LICENSE. Pro is unlocked by a development key, not by '
      + 'a ciscotools.dev subscription. Do not use this build in production.'
    : '';
}

async function refreshStatus(){renderStatus(await api('/api/status'));}
async function refreshLicense(){
  const out = await api('/api/license/refresh', {});
  msg('statusmsg', out.message || out.error, out.ok?'ok':'err');
  await refreshStatus();
}
async function doPlan(){
  const out = await api('/api/plan', {selection:selection(), commands:chosenCommands(), mode:'show'});
  msg('runmsg', out.error || (out.plan ? 'Plan: '+out.plan.device_count+' device(s), '+
      out.plan.command_count+' command(s).' : ''), out.error?'err':'ok');
  show('runout', JSON.stringify(out.plan, null, 2));
}
async function doRun(){
  const out = await api('/api/run', {selection:selection(), commands:chosenCommands(),
    mode:'show', dry_run: document.getElementById('dryrun').checked});
  msg('runmsg', out.message || out.error, out.ok?'ok':'err');
  show('runout', JSON.stringify(out.results || out.plan, null, 2));
}
async function exportSessions(){
  const sel = selection();
  const q = new URLSearchParams({group: sel.group||'', hostnames:(sel.hostnames||[]).join(',')});
  window.location = '/api/sessions.zip?' + q.toString();
}
async function doConfigPlan(){
  const out = await api('/api/plan', {selection:selection(), commands:[], mode:'config',
    template: document.getElementById('template').value,
    write_mem: document.getElementById('writemem').checked});
  msg('cfgmsg', out.error || 'Plan built.', out.error?'err':'ok');
  show('cfgout', JSON.stringify(out.plan || out, null, 2));
}
async function doConfigPush(){
  const dry = document.getElementById('cfgdry').checked;
  if(!dry && !confirm('This will change devices. Continue?')) return;
  const out = await api('/api/run', {selection:selection(), commands:[], mode:'config',
    template: document.getElementById('template').value,
    write_mem: document.getElementById('writemem').checked,
    dry_run: dry, confirm: !dry});
  msg('cfgmsg', out.message || out.error, out.ok?'ok':'err');
  show('cfgout', JSON.stringify(out.results || out.plan || out, null, 2));
}
async function captureBefore(){
  const out = await api('/api/before', {selection:selection(), commands:chosenCommands()});
  msg('cfgmsg', out.message || out.error, out.ok?'ok':'err');
}
async function diffAfter(){
  const out = await api('/api/diff', {selection:selection(), commands:chosenCommands()});
  msg('cfgmsg', out.message || out.error, out.ok?'ok':'err');
  show('cfgout', (out.diffs||[]).map(d=>d.diff||('# '+d.hostname+': no change')).join('\\n\\n'));
}
async function saveJob(){
  const out = await api('/api/jobs/save', {name: document.getElementById('jobname').value,
    spec: {group:selection().group, hostnames:selection().hostnames,
      commands: chosenCommands(), mode:'show', dry_run:true}});
  msg('jobmsg', out.message || out.error, out.ok?'ok':'err');
  await refreshStatus();
}
async function scheduleJob(){
  const out = await api('/api/jobs/schedule', {name: document.getElementById('jobname').value,
    schedule: document.getElementById('sched').value});
  msg('jobmsg', out.message || out.error, out.ok?'ok':'err');
  await refreshStatus();
}
refreshStatus();
</script></body></html>
"""


def parse_schedule_text(text):
    """'every 60m' / 'every 2h' / 'daily 02:30' -> a schedule dict or None."""
    cleaned = str(text or "").strip().lower()
    match = re.match(r"^every\s+(\d{1,4})\s*(m|min|mins|minutes|h|hr|hours)$", cleaned)
    if match:
        amount = int(match.group(1))
        if match.group(2).startswith("h"):
            amount *= 60
        return {"kind": "interval", "minutes": amount}
    match = re.match(r"^daily\s+(\S+)$", cleaned)
    if match and _TIME_OF_DAY_RE.match(match.group(1)):
        return {"kind": "daily", "at": match.group(1)}
    return None


class Handler(BaseHTTPRequestHandler):
    server_version = "%s/%s" % (APP_TITLE.replace(" ", ""), APP_VERSION)
    engine = None
    token = ""

    # -- plumbing -----------------------------------------------------------

    def log_message(self, fmt, *args):  # quieter, and never logs a body
        sys.stderr.write("[%s] %s %s\n" % (time.strftime("%H:%M:%S"),
                                           self.command, self.path.split("?")[0]))

    def _host_is_local(self):
        """Refuse a request whose Host header is not loopback.

        This is the DNS-rebinding guard: a page on the internet can make your
        browser send requests to 127.0.0.1, but it cannot control the Host
        header it sends, so requiring loopback there keeps a random web page
        from driving your network tooling.
        """
        host = (self.headers.get("Host") or "").rsplit(":", 1)[0].strip("[]")
        # An absent Host header is refused too: every browser and every curl
        # sends one, so a request without it is not a case worth accommodating.
        return host in ("127.0.0.1", "localhost", "::1")

    def _send(self, status, body, content_type="application/json",
              extra_headers=None):
        if isinstance(body, (dict, list)):
            body = json.dumps(body).encode("utf-8")
        elif isinstance(body, str):
            body = body.encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", content_type)
        self.send_header("Content-Length", str(len(body)))
        self.send_header("X-Content-Type-Options", "nosniff")
        self.send_header("Referrer-Policy", "no-referrer")
        self.send_header("Content-Security-Policy",
                         "default-src 'none'; style-src 'unsafe-inline'; "
                         "script-src 'unsafe-inline'; connect-src 'self'")
        for name, value in (extra_headers or {}).items():
            self.send_header(name, value)
        self.end_headers()
        if self.command != "HEAD":
            self.wfile.write(body)

    def _body(self):
        try:
            length = int(self.headers.get("Content-Length") or 0)
        except ValueError:
            return {}
        if length <= 0 or length > 4 * 1024 * 1024:
            return {}
        try:
            return json.loads(self.rfile.read(length).decode("utf-8"))
        except (ValueError, UnicodeError):
            return {}

    def _authorised(self):
        given = self.headers.get("X-NetOps-Token", "")
        return bool(self.token) and secrets.compare_digest(given, self.token)

    # -- routes -------------------------------------------------------------

    def do_GET(self):
        if not self._host_is_local():
            return self._send(403, {"ok": False, "error": "loopback only"})
        path, _, query = self.path.partition("?")
        params = urllib.parse.parse_qs(query)
        if path == "/":
            page = (PAGE_HTML.replace("__TITLE__", APP_TITLE)
                    .replace("__VERSION__", APP_VERSION)
                    .replace("__TOKEN__", self.token))
            return self._send(200, page, "text/html; charset=utf-8")
        if path == "/api/status":
            if not self._authorised():
                return self._send(403, {"ok": False, "error": "bad token"})
            return self._send(200, self.engine.status())
        if path == "/api/sessions.zip":
            if not self._authorised():
                return self._send(403, {"ok": False, "error": "bad token"})
            hostnames = [h for h in (params.get("hostnames", [""])[0] or "").split(",") if h]
            blob, count = self.engine.session_zip(
                {"group": params.get("group", [""])[0], "hostnames": hostnames or None})
            return self._send(200, blob, "application/zip", {
                "Content-Disposition": 'attachment; filename="securecrt-sessions.zip"',
                "X-Device-Count": str(count)})
        return self._send(404, {"ok": False, "error": "not found"})

    def do_POST(self):
        if not self._host_is_local():
            return self._send(403, {"ok": False, "error": "loopback only"})
        if not self._authorised():
            return self._send(403, {"ok": False, "error": "bad token"})
        path = self.path.partition("?")[0]
        body = self._body()
        engine = self.engine
        try:
            if path == "/api/license/refresh":
                ok, message = engine.license.refresh()
                return self._send(200, {"ok": ok,
                                        "message" if ok else "error": message})
            if path == "/api/plan":
                return self._send(200, {"ok": True, "plan": engine.plan(
                    body.get("selection") or {}, body.get("commands") or [],
                    mode=body.get("mode", "show"), template=body.get("template", ""),
                    write_mem=bool(body.get("write_mem")),
                    dry_run=body.get("dry_run"))})
            if path == "/api/run":
                return self._send(200, engine.run(
                    body.get("selection") or {}, body.get("commands") or [],
                    mode=body.get("mode", "show"), template=body.get("template", ""),
                    write_mem=bool(body.get("write_mem")),
                    dry_run=body.get("dry_run"), confirm=bool(body.get("confirm"))))
            if path == "/api/before":
                return self._send(200, engine.capture_before(
                    body.get("selection") or {}, body.get("commands") or []))
            if path == "/api/diff":
                return self._send(200, engine.diff_after(
                    body.get("selection") or {}, body.get("commands") or []))
            if path == "/api/jobs/save":
                return self._send(200, engine.save_job(body.get("name", ""),
                                                       body.get("spec") or {}))
            if path == "/api/jobs/schedule":
                gate = engine._require_pro()
                if gate:
                    return self._send(200, gate)
                schedule = parse_schedule_text(body.get("schedule", ""))
                if schedule is None:
                    return self._send(200, {"ok": False, "error":
                                            "Use 'every 60m', 'every 2h', or 'daily 02:30'."})
                job = engine.jobs.load(body.get("name", ""))
                if job is None:
                    return self._send(200, {"ok": False,
                                            "error": "Save the job first."})
                when, error = engine.scheduler.add(body.get("name", ""), job, schedule)
                if error:
                    return self._send(200, {"ok": False, "error": error})
                return self._send(200, {"ok": True,
                                        "message": "Scheduled. Next run %s." % _iso(when)})
        except Exception as error:  # noqa: BLE001 - never leak a traceback to the page
            return self._send(200, {"ok": False, "error": "%s" % error})
        return self._send(404, {"ok": False, "error": "not found"})


def serve(engine, host="127.0.0.1", port=8781, open_browser=True,
          token=None):  # pragma: no cover - exercised by hand, not in CI
    token = token or secrets.token_urlsafe(24)
    handler = type("BoundHandler", (Handler,),
                   {"engine": engine, "token": token})
    if host not in ("127.0.0.1", "localhost", "::1"):
        print("Refusing to bind %s: this app is loopback only." % host)
        host = "127.0.0.1"
    for candidate in range(port, port + 20):
        try:
            httpd = ThreadingHTTPServer((host, candidate), handler)
            break
        except OSError:
            continue
    else:
        print("Could not find a free port near %d." % port)
        return 1
    url = "http://%s:%d/" % (host, httpd.server_address[1])
    print("%s %s" % (APP_TITLE, APP_VERSION))
    print("  UI:       %s" % url)
    print("  Inventory %s (%d device(s))" % (engine.config.path("INVENTORY"),
                                             len(engine.devices)))
    print("  Dry run:  %s" % ("on" if engine.config.bool("DRY_RUN", True) else "OFF"))
    print("  License:  %s" % engine.license.state().reason)
    print("  Ctrl-C to stop. Nothing listens outside this machine.")
    if engine.pro():
        engine.scheduler.start()
    if open_browser:
        threading.Timer(0.4, lambda: webbrowser.open(url)).start()
    try:
        httpd.serve_forever()
    except KeyboardInterrupt:
        print("\nStopped.")
    finally:
        engine.scheduler.stop()
        httpd.server_close()
    return 0


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def _print_json(payload):
    print(json.dumps(payload, indent=2, sort_keys=True))


def main(argv=None):  # pragma: no cover - thin wrapper over tested pieces
    parser = argparse.ArgumentParser(
        prog=APP_NAME,
        description="%s %s - local config job runner. Runs on your machine."
                    % (APP_TITLE, APP_VERSION))
    parser.add_argument("--config", default="config.env",
                        help="path to config.env (default: next to this file)")
    parser.add_argument("--no-browser", action="store_true")
    parser.add_argument("--version", action="store_true")
    sub = parser.add_subparsers(dest="command")
    sub.add_parser("status", help="print status as JSON and exit")
    sub.add_parser("license", help="refresh the license and exit")
    plan_parser = sub.add_parser("plan", help="print a dry-run plan and exit")
    plan_parser.add_argument("--group", default="")
    plan_parser.add_argument("--command", action="append", default=[])
    export_parser = sub.add_parser("export-sessions",
                                   help="write a SecureCRT sessions zip and exit")
    export_parser.add_argument("--group", default="")
    export_parser.add_argument("--out", default="securecrt-sessions.zip")
    args = parser.parse_args(argv)

    if args.version:
        print("%s %s" % (APP_TITLE, APP_VERSION))
        return 0

    config_path = args.config
    if not os.path.isabs(config_path):
        here = os.path.dirname(os.path.abspath(__file__))
        candidate = os.path.join(here, config_path)
        config_path = candidate if os.path.exists(candidate) else os.path.abspath(config_path)
    config = Config.load(config_path)
    interactive = args.command is None and sys.stdin.isatty()
    config.prompt_for_missing_secrets(interactive=interactive)
    engine = Engine(config)

    if args.command == "status":
        _print_json(engine.status())
        return 0
    if args.command == "license":
        ok, message = engine.license.refresh()
        print(("OK: " if ok else "Not unlocked: ") + message)
        return 0 if ok else 1
    if args.command == "plan":
        _print_json(engine.plan({"group": args.group}, args.command))
        return 0
    if args.command == "export-sessions":
        blob, count = engine.session_zip({"group": args.group})
        with open(args.out, "wb") as handle:
            handle.write(blob)
        print("Wrote %s (%d device(s))." % (args.out, count))
        return 0

    return serve(engine, config.get("BIND_HOST", "127.0.0.1"),
                 config.int("BIND_PORT", 8781),
                 open_browser=config.bool("OPEN_BROWSER", True) and not args.no_browser)


if __name__ == "__main__":  # pragma: no cover
    sys.exit(main())