prole/config.py

"""
Shared configuration and utility functions for the Knoe installer (root-level).

This mirrors `knoe.knoe.config` but is located under the root `installer/`
package per the refactor request. UI code should import from `knoe.config`.
"""

from __future__ import annotations

import base64
import getpass
import json
import os
import platform
import re
import shlex
import shutil
import subprocess
import sys
import urllib.request
from pathlib import Path
from typing import Optional, Tuple

from cryptography.hazmat.primitives.ciphers.aead import AESGCM


def resolve_knoe_home(env: dict[str, str] | None = None) -> Path:
    """Resolve `KNOE_HOME` from `env`/process environment.

    Falls back to `$HOME/.knoe` when `KNOE_HOME` is not set.

    NOTE: This is duplicated here (and in `knoe.core.env`) to avoid circular
    imports between the two modules.
    """

    env_map = env or os.environ
    raw = (env_map.get("KNOE_HOME") or "").strip()
    if raw:
        expanded = os.path.expanduser(os.path.expandvars(raw))
        return Path(expanded)
    return Path.home() / ".knoe"

# Secret handling (temporary encrypted values in knoe.cfg)
KNOE_SECRET_PREFIX = "${KNOE_SECRET:"
KNOE_SECRET_SUFFIX = "}"
OPENBAO_PREFIX = "${OPENBAO:"
OPENBAO_SUFFIX = "}"
KNOE_SECRET_VERSION = "v1"
KNOE_SECRET_SERVICE = "knoe-installer"

# Map config keys to OpenBao paths (namespace injected at runtime)
SECRET_KEY_SPECS = {
    ("Inputs", "init_password.db_password"): ("db", "password"),
    ("Inputs", "init_password.db_password_confirm"): ("db", "password"),
    ("Inputs", "kerberos_config.password"): ("kerberos", "password"),
    ("Global", "DB_PASSWORD"): ("db", "password"),
    ("Kerberos Authentication", "PASSWORD"): ("kerberos", "password"),
    ("Monitoring", "GRAFANA_ADMIN_PASSWORD"): ("monitoring", "grafana_admin_password"),
}


def _is_knoe_secret(value: str | None) -> bool:
    return (
        bool(value)
        and value.startswith(KNOE_SECRET_PREFIX)
        and value.endswith(KNOE_SECRET_SUFFIX)
    )


def _is_openbao_ref(value: str | None) -> bool:
    return (
        bool(value)
        and value.startswith(OPENBAO_PREFIX)
        and value.endswith(OPENBAO_SUFFIX)
    )


def _get_secret_key_file() -> Path:
    return resolve_knoe_home() / "secrets" / "knoe.key"


def _get_keychain_key(service: str, account: str) -> bytes:
    try:
        res = subprocess.run(
            ["security", "find-generic-password", "-a", account, "-s", service, "-w"],
            capture_output=True,
            text=True,
        )
        if res.returncode == 0 and res.stdout.strip():
            return base64.urlsafe_b64decode(res.stdout.strip().encode("utf-8"))
    except Exception:
        pass

    key_b64 = base64.urlsafe_b64encode(os.urandom(32)).decode("utf-8")
    try:
        subprocess.run(
            [
                "security",
                "add-generic-password",
                "-a",
                account,
                "-s",
                service,
                "-w",
                key_b64,
                "-U",
            ],
            capture_output=True,
            text=True,
        )
    except Exception:
        pass
    return base64.urlsafe_b64decode(key_b64.encode("utf-8"))


def _get_file_key(path: Path) -> bytes:
    path.parent.mkdir(parents=True, exist_ok=True)
    if path.exists():
        raw = path.read_text().strip()
        try:
            return base64.urlsafe_b64decode(raw.encode("utf-8"))
        except Exception:
            pass
    key_b64 = base64.urlsafe_b64encode(os.urandom(32)).decode("utf-8")
    path.write_text(key_b64)
    try:
        os.chmod(path, 0o600)
    except Exception:
        pass
    return base64.urlsafe_b64decode(key_b64.encode("utf-8"))


def _get_secret_key() -> bytes:
    system = platform.system()
    account = getpass.getuser() or "knoe"
    if system == "Darwin":
        return _get_keychain_key(KNOE_SECRET_SERVICE, account)
    return _get_file_key(_get_secret_key_file())


def _encrypt_knoe_secret(plaintext: str) -> str:
    if plaintext is None:
        return ""
    if _is_knoe_secret(plaintext):
        return plaintext
    key = _get_secret_key()
    aesgcm = AESGCM(key)
    nonce = os.urandom(12)
    ciphertext = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
    nonce_b64 = base64.urlsafe_b64encode(nonce).decode("utf-8")
    ct_b64 = base64.urlsafe_b64encode(ciphertext).decode("utf-8")
    return f"{KNOE_SECRET_PREFIX}{KNOE_SECRET_VERSION}:{nonce_b64}:{ct_b64}{KNOE_SECRET_SUFFIX}"


def _decrypt_knoe_secret(value: str) -> str:
    if not _is_knoe_secret(value):
        return value
    inner = value[len(KNOE_SECRET_PREFIX) : -len(KNOE_SECRET_SUFFIX)]
    parts = inner.split(":")
    if len(parts) != 3 or parts[0] != KNOE_SECRET_VERSION:
        return value
    try:
        nonce = base64.urlsafe_b64decode(parts[1].encode("utf-8"))
        ciphertext = base64.urlsafe_b64decode(parts[2].encode("utf-8"))
        key = _get_secret_key()
        aesgcm = AESGCM(key)
        plaintext = aesgcm.decrypt(nonce, ciphertext, None)
        return plaintext.decode("utf-8")
    except Exception:
        return value


def _encrypt_cfg_secret(plaintext: str | None) -> str:
    if not plaintext:
        return ""
    if _is_knoe_secret(plaintext) or _is_openbao_ref(plaintext):
        return plaintext
    try:
        return _encrypt_knoe_secret(plaintext)
    except Exception:
        return str(plaintext)


def _resolve_openbao_ref(value: str) -> str:
    if not _is_openbao_ref(value):
        return value
    inner = value[len(OPENBAO_PREFIX) : -len(OPENBAO_SUFFIX)]
    if ":" not in inner:
        return value
    path, key = inner.split(":", 1)
    if not path or not key:
        return ""
    mount = "kv"
    secret_path = path
    if "/" in path:
        maybe_mount, rest = path.split("/", 1)
        if maybe_mount:
            mount = maybe_mount
            secret_path = rest
    token = os.environ.get("OPENBAO_ROOT_TOKEN", "")
    if not token:
        knoe_service = os.environ.get("KNOE_SERVICE")
        if knoe_service:
            token_path = Path(knoe_service) / "secrets" / "openbao-root-token"
            if token_path.exists():
                token = token_path.read_text().strip()
    if not token:
        return value
    url = os.environ.get("PROLE_OPENBAO_URL")
    if not url:
        for p in ["8200", "18200"]:
            try:
                with urllib.request.urlopen(
                    f"http://127.0.0.1:{p}/v1/sys/health", timeout=0.5
                ) as r:
                    if r.getcode() == 200:
                        url = f"http://127.0.0.1:{p}"
                        break
            except Exception:
                pass
    if not url:
        url = "http://127.0.0.1:8200"
    url = url.rstrip("/")
    try:
        req = urllib.request.Request(f"{url}/v1/{mount}/data/{secret_path}")
        req.add_header("X-Vault-Token", token)
        with urllib.request.urlopen(req, timeout=4) as resp:
            payload = json.loads(resp.read().decode("utf-8"))
        return payload.get("data", {}).get("data", {}).get(key, "") or ""
    except Exception:
        return value


def _resolve_secret_value(value: str) -> str:
    if _is_knoe_secret(value):
        return _decrypt_knoe_secret(value)
    if _is_openbao_ref(value):
        return _resolve_openbao_ref(value)
    return value


# .properties loader (simple key=value, # comments)
def _load_properties(path: Path) -> dict:
    props: dict[str, str] = {}
    try:
        text = path.read_text(encoding="utf-8")
    except Exception:
        return props
    for line in text.splitlines():
        s = line.strip()
        if not s or s.startswith("#"):
            continue
        if "=" in s:
            k, v = s.split("=", 1)
            k = k.strip()
            v = v.strip()
            if k:
                props[k] = v
    return props


# Repository root: this file lives at <repo>/installer/config.py → parent is repo
PROJECT_ROOT = Path(__file__).resolve().parents[1]
KNOE_APP_DIR = PROJECT_ROOT / "knoe-app"
KNOE_PROPS_PATH = KNOE_APP_DIR / "knoe.properties"

import logging


def setup_logging(verbose=False, debug=False):
    """Set up logging for the installer.

    Levels:
      default  – WARNING on console (quiet)
      verbose  – INFO on console
      debug    – DEBUG to a datestamped file *and* INFO on console
    """
    if debug:
        console_level = logging.INFO
        file_level = logging.DEBUG
    elif verbose:
        console_level = logging.INFO
        file_level = None
    else:
        console_level = logging.WARNING
        file_level = None

    console_format = "%(asctime)s [%(levelname)s] %(name)s: %(message)s"
    debug_format = "%(asctime)s.%(msecs)03d [%(levelname)-5s] %(name)s (%(filename)s:%(lineno)d) %(funcName)s: %(message)s"

    console_handler = logging.StreamHandler()
    console_handler.setLevel(console_level)
    console_handler.setFormatter(logging.Formatter(console_format, datefmt="%H:%M:%S"))

    handlers: list[logging.Handler] = [console_handler]

    if debug and file_level is not None:
        log_dir = PROJECT_ROOT / "logs"
        log_dir.mkdir(parents=True, exist_ok=True)
        from datetime import datetime

        ts = datetime.now().strftime("%Y%m%d-%H%M%S")
        debug_log = log_dir / f"install-debug-{ts}.log"
        file_handler = logging.FileHandler(debug_log)
        file_handler.setLevel(file_level)
        file_handler.setFormatter(
            logging.Formatter(debug_format, datefmt="%Y-%m-%d %H:%M:%S")
        )
        handlers.append(file_handler)
        print(f"Debug logging enabled to {debug_log}")

    logging.basicConfig(
        level=logging.DEBUG if debug else console_level, handlers=handlers, force=True
    )


_KNOE_PROPS_CACHE: Optional[dict] = None


def get_properties() -> dict:
    """Load and cache knoe.properties from the repo (installer context).

    Order:
    - Repo default at knoe-app/knoe.properties
    - Optional env override: KNOE_PROPERTIES points to a file
    """
    global _KNOE_PROPS_CACHE
    if _KNOE_PROPS_CACHE is None:
        props: dict[str, str] = {}
        # repo default
        if KNOE_PROPS_PATH.exists():
            props.update(_load_properties(KNOE_PROPS_PATH))
        # env override (absolute path)
        env_path = os.environ.get("KNOE_PROPERTIES")
        if env_path:
            p = Path(env_path)
            if p.exists():
                props.update(_load_properties(p))
        _KNOE_PROPS_CACHE = props
    return dict(_KNOE_PROPS_CACHE)


def get_config_value(key: str, default: Optional[str] = None) -> Optional[str]:
    return get_properties().get(key, default)


def get_ui_icon_image_path() -> Path:
    """Return absolute path to the UI icon image for the installer.

    Priority (new → legacy):
    - `icon` key in knoe.properties (requested)
    - legacy `ui.icon`

    Defaults to img/knoeIcon.png under repo root if not set or missing.
    """
    # Prefer new key `icon`, fall back to old `ui.icon`
    rel = get_config_value("icon") or get_config_value("ui.icon") or "img/knoeIcon.png"
    p = (PROJECT_ROOT / rel).resolve()
    if p.exists():
        return p
    # fallback
    return (PROJECT_ROOT / "img/knoeIcon.png").resolve()


def get_ui_background_image_path() -> Path:
    """Return absolute path to the UI background image.

    Priority (new → legacy):
    - `background` key in knoe.properties (requested)
    - legacy `ui.background`

    Defaults to img/knoeLogoSepia.png under repo root if not set or missing.
    """
    rel = (
        get_config_value("background")
        or get_config_value("ui.background")
        or "img/knoeLogoSepia.png"
    )
    p = (PROJECT_ROOT / rel).resolve()
    if p.exists():
        return p
    # fallback
    return (PROJECT_ROOT / "img/knoeLogoSepia.png").resolve()


def is_apple_silicon() -> bool:
    return platform.machine() == "arm64" and platform.system() == "Darwin"


def get_docker_build_platform_args(target_env: Optional[str] = None) -> list[str]:
    override = os.environ.get("PROLE_DOCKER_PLATFORM", "").strip()
    if override:
        return ["--platform", override]
    env_key = (target_env or "").strip().lower()
    if env_key == "dev" and is_apple_silicon():
        return ["--platform", "linux/arm64"]
    if env_key in ("service", "k3s", "knoe-service-cluster"):
        return ["--platform", "linux/arm64"]
    if is_apple_silicon():
        return ["--platform", "linux/amd64"]
    return []


def normalize_version(text: str) -> str:
    if not text:
        return ""
    m = re.search(r"(\d+(?:[._-]\d+){0,5})", text)
    if not m:
        nums = re.findall(r"\d+", text)
    else:
        nums = re.findall(r"\d+", m.group(1))
    if not nums:
        return text.strip()
    parts = (nums + ["0", "0"])[:3]
    try:
        parts = [f"{int(p):02d}" for p in parts]
    except Exception:
        parts = [p.zfill(2) for p in parts]
    return ".".join(parts)


def get_resource_path(relative_path: str | Path) -> Path:
    """Get absolute path to resource, works for dev and for PyInstaller."""
    try:
        # PyInstaller creates a temp folder and stores path in _MEIPASS
        base_path = sys._MEIPASS
    except Exception:
        base_path = PROJECT_ROOT

    return Path(base_path) / relative_path


def _expand_path(val: str | None) -> str:
    if not val:
        return ""
    # Expand ~ and shell-style variables like $HOME. Use a small fixed-point loop
    # so nested refs (e.g. KNOE_HOME=$HOME/...) expand fully.
    out = os.path.expanduser(str(val))
    for _ in range(10):
        new = os.path.expandvars(out)
        if new == out:
            break
        out = new
    return out


def _normalize_persistence_mode(mode: str | None) -> str:
    raw = (mode or "").strip().lower()
    if raw in {"dev", "k3d", "local"}:
        return "dev"
    if raw in {"service", "k3s", "k3s-hosts", "knoe-service-cluster"}:
        return "k3s-hosts"
    if raw in {"prod", "k8s", "knoe-prod-cluster"}:
        return "prod"
    return raw


def _is_simple_template_placeholder(value: str | None) -> bool:
    v = (value or "").strip()
    return bool(re.match(r"^\$\{[A-Za-z_][A-Za-z0-9_]*\}$", v))


def _looks_like_localhost_value(value: str | None) -> bool:
    v = (value or "").strip().lower()
    if not v:
        return False
    return "localhost" in v or "127.0.0.1" in v


def _looks_like_generated_service_endpoint_key(key: str | None) -> bool:
    k = (key or "").strip().upper()
    if not k:
        return False
    return (
        k == "SERVICE_HOSTNAME"
        or k.endswith("_SERVICE_ENDPOINT")
        or k.endswith("_SERVICE_HOSTNAME")
        or (k.startswith("SERVICE_") and k.endswith("_HOSTNAME"))
    )


def _normalize_cfg_value_for_persistence(
    section: str,
    key: str,
    value: str | None,
    *,
    mode: str | None = None,
    explicit: bool = False,
) -> str | None:
    """Return a persistable value, or `None` if it must not be persisted.

    Policy:
    - Dev/k3d: retain existing behavior.
    - k3s-hosts/Service and Prod: keep explicit inputs/handoff values, but drop
      derived placeholders, generated endpoint hostnames, localhost-based
      derived values, and k3d/dev endpoint assumptions.
    """

    raw_value = "" if value is None else str(value)
    trimmed = raw_value.strip()
    if not trimmed:
        return None

    normalized_mode = _normalize_persistence_mode(mode)
    if normalized_mode not in {"k3s-hosts", "prod"}:
        return trimmed

    normalized_key = (key or "").strip().upper()
    if _is_simple_template_placeholder(trimmed):
        return None

    if normalized_key == "SERVICE_NAMESPACE" and not explicit:
        return None

    if _looks_like_generated_service_endpoint_key(normalized_key) and not explicit:
        return None

    if _looks_like_localhost_value(trimmed) and not explicit:
        return None

    if "k3d.localhost" in trimmed.lower() and not explicit:
        return None

    return trimmed


def _filter_cfg_values_for_persistence(
    section: str,
    values: dict | None,
    *,
    mode: str | None = None,
    explicit_keys: set[str] | None = None,
) -> dict[str, str]:
    """Filter section values using mode-aware persistence rules."""

    if not values or not isinstance(values, dict):
        return {}

    explicit_keys_upper = {str(k).strip().upper() for k in (explicit_keys or set())}
    filtered: dict[str, str] = {}
    for key, value in values.items():
        key_str = str(key)
        normalized = _normalize_cfg_value_for_persistence(
            section,
            key_str,
            "" if value is None else str(value),
            mode=mode,
            explicit=key_str.strip().upper() in explicit_keys_upper,
        )
        if normalized is None:
            continue
        filtered[key_str] = normalized

    return filtered


def _collect_cfg_vars(
    cfg: any,
    *,
    mode: str | None = None,
    explicit_keys: set[tuple[str, str]] | None = None,
) -> dict:
    variables: dict[str, str] = {}
    explicit_pairs = {
        (str(sec).strip().upper(), str(key).strip().upper())
        for sec, key in (explicit_keys or set())
    }

    def _is_placeholder(v: str) -> bool:
        vv = (v or "").strip()
        return vv.startswith("${") and vv.endswith("}")

    def _merge_section(section: str) -> None:
        if not cfg.has_section(section):
            return
        for k, v in cfg.items(section):
            s = str(v or "").strip()
            # Avoid polluting the variable map with empty/self-referencing placeholders.
            if not s or _is_placeholder(s):
                continue
            if mode:
                norm = _normalize_cfg_value_for_persistence(
                    section,
                    str(k),
                    s,
                    mode=mode,
                    explicit=(section.upper(), str(k).strip().upper()) in explicit_pairs,
                )
                if norm is None:
                    continue
                s = norm
            variables[k] = s

    # Global is the baseline. System and User can override, but only with concrete values.
    _merge_section("Global")
    _merge_section("System Environment")
    _merge_section("User")
    return variables


def _collect_cfg_vars_from_data(
    cfg_data: dict | None,
    *,
    mode: str | None = None,
    explicit_keys: set[tuple[str, str]] | None = None,
) -> dict[str, str]:
    """Collect variable names from a `knoe_cfg_data`-style dict.

    This mirrors `_collect_cfg_vars()` but operates on the UI/controller's in-memory
    dict instead of a `configparser.ConfigParser`.
    """
    variables: dict[str, str] = {}
    if not cfg_data or not isinstance(cfg_data, dict):
        return variables
    explicit_pairs = {
        (str(sec).strip().upper(), str(key).strip().upper())
        for sec, key in (explicit_keys or set())
    }

    def _is_placeholder(v: str) -> bool:
        vv = (v or "").strip()
        return vv.startswith("${") and vv.endswith("}")

    for section in ("Global", "System Environment", "User"):
        sec = cfg_data.get(section, {}) or {}
        if not isinstance(sec, dict):
            continue
        for k, v in sec.items():
            s = str(v or "").strip()
            if not s or _is_placeholder(s):
                continue
            if mode:
                norm = _normalize_cfg_value_for_persistence(
                    section,
                    str(k),
                    s,
                    mode=mode,
                    explicit=(section.upper(), str(k).strip().upper()) in explicit_pairs,
                )
                if norm is None:
                    continue
                s = norm
            variables[str(k)] = s
    return variables


def _expand_cfg_vars_shellstyle(val: str, variables: dict[str, str]) -> str:
    """Expand `$VAR` and `${VAR}` using `variables`.

    Notes:
    - Only expands simple shell-style identifiers (letters/digits/underscore).
      This intentionally avoids treating `${KNOE_SECRET:...}` or similar
      colon-delimited references as variables.
    - Unknown variables are left intact.
    """
    if not val or not isinstance(val, str) or not variables:
        return val

    # ${VAR}
    def repl_braced(m: re.Match) -> str:
        name = m.group(1)
        return variables.get(name, m.group(0))

    out = re.sub(r"\$\{([A-Za-z_][A-Za-z0-9_]*)\}", repl_braced, val)

    # $VAR (avoid $ followed by { ... } which is already handled above)
    def repl_plain(m: re.Match) -> str:
        name = m.group(1)
        return variables.get(name, m.group(0))

    out = re.sub(r"\$([A-Za-z_][A-Za-z0-9_]*)", repl_plain, out)
    return out


def _expand_path_expr(val: str | None, variables: dict[str, str] | None = None) -> str:
    """Expand a path-like expression with support for config variables.

    Expansion order (fixed-point loop):
    1) `~` expansion
    2) config variables (`$VAR` and `${VAR}`)
    3) OS environment variables (via `os.path.expandvars`)
    """
    if not val:
        return ""

    out = os.path.expanduser(str(val))
    for _ in range(10):
        new = out
        if variables:
            new = _expand_cfg_vars_shellstyle(new, variables)
        new = os.path.expandvars(new)
        if new == out:
            break
        out = new
    return out


def _expand_cfg_value(val: str, variables: dict) -> str:
    if not val or not isinstance(val, str):
        return val
    # Basic ${VAR} expansion
    import re

    result = val
    for m in re.finditer(r"\$\{([^}]+)\}", val):
        full, var = m.group(0), m.group(1)
        if var in variables:
            result = result.replace(full, variables[var])
    return result


def _parse_bool(val: any, default: bool | None = False) -> bool | None:
    if val is None:
        return default
    if isinstance(val, bool):
        return val
    s = str(val).lower().strip()
    if s in ("true", "1", "yes", "on"):
        return True
    if s in ("false", "0", "no", "off"):
        return False
    return default


def _extract_yaml_scalar_from_text(text: str, key: str) -> str:
    # Look for key: "value" or key: value
    m = re.search(rf'^{key}:\s*["\']?(.*?)["\']?\s*$', text, re.MULTILINE)
    if m:
        return m.group(1)
    return ""


def _extract_inline_vault_block(text: str, key: str) -> str:
    lines = text.splitlines()
    for i, line in enumerate(lines):
        if line.strip().startswith(f"{key}:"):
            base_indent = len(line) - len(line.lstrip())
            block = []
            i += 1
            while i < len(lines):
                line = lines[i]
                if not line.strip():
                    i += 1
                    continue
                indent = len(line) - len(line.lstrip())
                if indent <= base_indent:
                    break
                block.append(line.strip())
                i += 1
            if block and block[0].startswith("$ANSIBLE_VAULT"):
                return "\n".join(block) + "\n"
            return ""
    return ""


def _try_read_ansible_vault_value(vault_path: Path, key: str) -> str:
    if not vault_path.exists():
        return ""

    # Simple search in plain text first
    try:
        raw_text = vault_path.read_text()
        plain = _extract_yaml_scalar_from_text(raw_text, key)
        if (
            plain
            and not plain.startswith("$ANSIBLE_VAULT")
            and not plain.startswith("!vault")
        ):
            return plain
    except Exception:
        pass

    password_file = (os.environ.get("ANSIBLE_VAULT_PASSWORD_FILE") or "").strip()
    if not password_file:
        candidate = PROJECT_ROOT / ".vault_pass"
        if candidate.is_file():
            password_file = str(candidate)

    if not password_file or shutil.which("ansible-vault") is None:
        return ""

    def _vault_view(path: str) -> str:
        try:
            res = subprocess.run(
                ["ansible-vault", "view", path, "--vault-password-file", password_file],
                capture_output=True,
                text=True,
                timeout=5,
                env=os.environ.copy(),
                stdin=subprocess.DEVNULL,
            )
            if res.returncode == 0:
                return res.stdout or ""
        except Exception:
            pass
        return ""

    try:
        output = _vault_view(str(vault_path))
        if output:
            return _extract_yaml_scalar_from_text(output, key)

        # Inline vault
        inline_block = _extract_inline_vault_block(vault_path.read_text(), key)
        if not inline_block:
            return ""

        import tempfile

        with tempfile.NamedTemporaryFile(mode="w", delete=False) as tmp:
            tmp.write(inline_block)
            tmp_name = tmp.name

        try:
            output = _vault_view(tmp_name)
            return output.strip()
        finally:
            if os.path.exists(tmp_name):
                os.unlink(tmp_name)
    except Exception:
        pass
    return ""


def _update_knoe_cfg_value(
    section: str,
    key: str,
    value: str,
    *,
    mode: str | None = None,
    explicit: bool = False,
):
    # Prefer `$KNOE_CONF/knoe.cfg` (single entrypoint) and follow symlink so we
    # update the active environment base file without mutating other environments.
    try:
        from knoe import knoe_conf

        conf_dir = knoe_conf.resolve_knoe_conf_dir(PROJECT_ROOT)
        cfg_path = knoe_conf.entrypoint_path(conf_dir)
    except Exception:
        cfg_path = PROJECT_ROOT / "conf" / "knoe.cfg"

    if not cfg_path.exists():
        return

    try:
        write_path = cfg_path.resolve() if cfg_path.is_symlink() else cfg_path
    except Exception:
        write_path = cfg_path

    if not write_path.exists():
        return

    import configparser

    cfg = configparser.ConfigParser(interpolation=None)
    cfg.optionxform = str
    cfg.read(write_path)

    if not cfg.has_section(section):
        cfg.add_section(section)

    effective_mode = mode
    if effective_mode is None:
        effective_mode = (
            os.environ.get("DEPLOYMENT_MODE")
            or os.environ.get("CLUSTER_ENV")
            or os.environ.get("MODE")
            or ""
        )

    normalized_value = _normalize_cfg_value_for_persistence(
        section,
        key,
        value,
        mode=effective_mode,
        explicit=explicit,
    )

    if normalized_value is None:
        if cfg.has_option(section, key):
            cfg.remove_option(section, key)
        with open(write_path, "w") as f:
            cfg.write(f)
        return

    cfg.set(section, key, normalized_value)

    with open(write_path, "w") as f:
        cfg.write(f)


def _write_k3s_kubeconfig(server_url: str, token: str) -> Path:
    if not server_url.startswith("http"):
        server_url = f"https://{server_url}"
    cfg = (
        "apiVersion: v1\n"
        "kind: Config\n"
        "clusters:\n"
        "- cluster:\n"
        f"    server: {server_url}\n"
        "    insecure-skip-tls-verify: true\n"
        "  name: knoe-k3s\n"
        "contexts:\n"
        "- context:\n"
        "    cluster: knoe-k3s\n"
        "    user: knoe-k3s\n"
        "  name: knoe-k3s\n"
        "current-context: knoe-k3s\n"
        "users:\n"
        "- name: knoe-k3s\n"
        "  user:\n"
        f"    token: {token}\n"
    )
    knoe_service = os.environ.get("KNOE_SERVICE", "").strip()
    if knoe_service:
        path = Path(knoe_service).expanduser() / "secrets" / "k3s.kubeconfig"
    else:
        knoe_home = os.environ.get("KNOE_HOME", "").strip()
        if knoe_home:
            path = Path(knoe_home).expanduser() / "knoe-k3s.kubeconfig"
        else:
            path = PROJECT_ROOT / "knoe-k3s.kubeconfig"
    # Never overwrite a kubeconfig that uses client-certificate auth (e.g.
    # one fetched by Ansible) with a token-based fallback.
    if path.exists():
        try:
            existing = path.read_text(encoding="utf-8")
            if "client-certificate-data" in existing:
                # Keep the certificate-based kubeconfig intact, but still generate a
                # token-based kubeconfig alongside it so callers/tests get a deterministic
                # token config for the requested server.
                path = path.with_name(f"{path.stem}.token{path.suffix}")
        except Exception:
            pass
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(cfg, encoding="utf-8")
    os.chmod(path, 0o600)
    return path


def _merge_kubeconfig(kubeconfig_path: str) -> bool:
    """Merge a standalone kubeconfig into ~/.kube/config so kubectx can discover it."""
    default_kube = Path.home() / ".kube" / "config"
    default_kube.parent.mkdir(parents=True, exist_ok=True)

    merged_env = os.environ.copy()
    paths = []
    if default_kube.exists():
        paths.append(str(default_kube))
    paths.append(str(kubeconfig_path))
    merged_env["KUBECONFIG"] = os.pathsep.join(paths)

    try:
        res = subprocess.run(
            ["kubectl", "config", "view", "--flatten"],
            capture_output=True,
            text=True,
            env=merged_env,
            timeout=10,
        )
        if res.returncode == 0 and res.stdout.strip():
            import tempfile

            fd, tmp = tempfile.mkstemp(dir=str(default_kube.parent), suffix=".tmp")
            try:
                os.write(fd, res.stdout.encode("utf-8"))
                os.close(fd)
                fd = -1
                os.replace(tmp, str(default_kube))
                os.chmod(str(default_kube), 0o600)
                return True
            except Exception:
                if fd >= 0:
                    os.close(fd)
                try:
                    os.unlink(tmp)
                except Exception:
                    pass
    except Exception:
        pass
    return False


DEPENDENCIES = [
    {
        "id": "brew",
        "name": "Homebrew",
        "description": "Package manager for macOS",
        "url": "https://brew.sh",
        "install_cmd": '/bin/bash -c "$(curl -fsSL https://raw.githubusercontent.com/Homebrew/install/HEAD/install.sh)"',
        "check_cmd": "brew --version",
        "bin": "brew",
    },
    {
        "id": "python",
        "name": "python",
        "parent": "brew",
        "description": "Python programming language",
        "url": "https://www.python.org",
        "install_cmd": "brew install python",
        "check_cmd": "python3 --version",
        "bin": "python3",
    },
    {
        "id": "ansible",
        "name": "ansible",
        "parent": "brew",
        "description": "Infrastructure automation tool",
        "url": "https://www.ansible.com",
        "install_cmd": "brew install ansible",
        "check_cmd": "ansible --version",
        "bin": "ansible",
    },
    {
        "id": "kubectl",
        "name": "kubectl",
        "parent": "brew",
        "description": "Kubernetes command-line tool",
        "url": "https://kubernetes.io/docs/reference/kubectl/",
        "install_cmd": "brew install kubectl",
        "check_cmd": "kubectl version --client",
        "bin": "kubectl",
    },
    {
        "id": "kubectx",
        "name": "kubectx",
        "parent": "brew",
        "description": "Tool to switch between Kubernetes contexts",
        "url": "https://github.com/ahmetb/kubectx",
        "install_cmd": "brew install kubectx",
        "check_cmd": "kubectx -h",
        "version_cmd": "brew list --versions kubectx 2>/dev/null || brew info kubectx 2>/dev/null | head -n1",
        "bin": "kubectx",
    },
    {
        "id": "docker",
        "name": "Docker",
        "description": "Container platform for running Knoe services",
        "url": "https://www.docker.com/products/docker-desktop",
        "install_cmd": None,
        "check_cmd": "docker --version",
        "bin": "docker",
    },
    {
        "id": "k3d",
        "name": "k3d",
        "description": "Lightweight wrapper to run k3s in Docker",
        "url": "https://k3d.io",
        "install_cmd": "brew install k3d",
        "check_cmd": "k3d --version",
        "bin": "k3d",
    },
    {
        "id": "opentofu",
        "name": "OpenTofu",
        "parent": "brew",
        "description": "Open-source infrastructure as code tool",
        "url": "https://opentofu.org",
        "install_cmd": "brew install opentofu",
        "check_cmd": "tofu --version",
        "bin": "tofu",
    },
]


def _resolve_brew_bin() -> str | None:
    """Return an absolute path to a `brew` binary if one is found in common locations."""

    candidates: list[str] = [
        "/home/linuxbrew/.linuxbrew/bin/brew",  # Linuxbrew (recommended)
        os.path.expanduser("~/.linuxbrew/bin/brew"),  # Linuxbrew (legacy single-user)
        "/opt/homebrew/bin/brew",  # macOS (Apple Silicon)
        "/usr/local/bin/brew",  # macOS (Intel)
    ]

    for p in candidates:
        try:
            if os.path.isfile(p) and os.access(p, os.X_OK):
                return p
        except Exception:
            continue
    return None


def _augment_env_for_brew(env: dict | None = None) -> dict:
    """Prepend Homebrew `bin`/`sbin` to PATH when brew exists but isn't in PATH.

    This keeps dependency checks and `brew install ...` invocations working in
    non-interactive environments where shell init files aren't sourced.
    """

    if env is None:
        env = os.environ.copy()

    brew_bin = _resolve_brew_bin()
    if not brew_bin:
        return env

    try:
        brew_bin_path = Path(brew_bin)
        prefix = brew_bin_path.parent.parent
        bin_dir = str(prefix / "bin")
        sbin_dir = str(prefix / "sbin")

        path_val = env.get("PATH", "") or ""
        parts = [p for p in path_val.split(os.pathsep) if p]

        prepend: list[str] = []
        if bin_dir and bin_dir not in parts:
            prepend.append(bin_dir)
        try:
            if os.path.isdir(sbin_dir) and sbin_dir not in parts:
                prepend.append(sbin_dir)
        except Exception:
            # sbin is optional
            pass

        if prepend:
            env["PATH"] = os.pathsep.join(prepend + parts)
    except Exception:
        return env

    return env


def get_dep_info(dep: dict) -> Tuple[bool, Optional[str], Optional[str]]:
    bin_name = dep.get("bin") or dep["name"]
    location = None
    version = None
    installed = False
    try:
        env = _augment_env_for_brew(os.environ.copy())
        if bin_name:
            # Avoid `bash -l` here: login shell init scripts on some distros
            # can overwrite PATH, defeating our augmented PATH for Linuxbrew.
            try:
                location = shutil.which(bin_name, path=env.get("PATH"))
            except Exception:
                location = None

            # Extra safety: if checking for brew specifically, fall back to
            # well-known absolute locations even if PATH lookup fails.
            if not location and bin_name == "brew":
                location = _resolve_brew_bin()

            if location:
                installed = True

        check_cmd = dep.get("check_cmd")
        if check_cmd:
            res2 = subprocess.run(
                ["bash", "-c", check_cmd],
                capture_output=True,
                text=True,
                env=env,
            )
            if res2.returncode == 0:
                installed = True
                # Use first line as version if not already set
                if not version:
                    version = " ".join(res2.stdout.strip().splitlines()[:1])
                if not location:
                    location = "Installed via Python"

        version_cmd = dep.get("version_cmd")
        if version_cmd:
            res3 = subprocess.run(
                ["bash", "-c", version_cmd],
                capture_output=True,
                text=True,
                env=env,
            )
            if res3.returncode == 0:
                v_out = res3.stdout.strip()
                if v_out:
                    version = " ".join(v_out.splitlines()[:1])

    except Exception:
        pass
    return installed, location, version


def load_dependencies(refresh: bool = True) -> list[dict]:
    """Return a list of dependency descriptors.

    Each item mirrors entries from ``DEPENDENCIES`` and is enriched with:
    - installed: bool
    - location: Optional[str]
    - version: Optional[str]
    """
    deps: list[dict] = [dict(d) for d in DEPENDENCIES]
    if refresh:
        for dep in deps:
            try:
                installed, location, version = get_dep_info(dep)
                dep["installed"] = bool(installed)
                if location is not None:
                    dep["location"] = location
                if version is not None:
                    dep["version"] = version
            except Exception:
                dep["installed"] = False
    return deps