prole/tests/silent_install_test.sh

#!/bin/bash

set -euo pipefail

ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"

# Logging setup
LOG_DIR="${PROLE_LOGS:-$ROOT_DIR/logs}"
LOG_DIR="${LOG_DIR%/}"
mkdir -p "$LOG_DIR"
TIMESTAMP="$(date +%Y%m%d-%H%M%S)"
LOG_FILE="${LOG_DIR}/silent_install_${TIMESTAMP}.log"

# Function to handle output redirection
if [[ "${SILENT_INSTALL_LOG:-}" == "true" ]]; then
  echo "Logging output to: $LOG_FILE"
  # Redirect stdout and stderr to both terminal and log file
  exec > >(tee -a "$LOG_FILE") 2>&1
fi

CFG_PATH="${1:-}"
if [[ -z "$CFG_PATH" ]]; then
  if [[ -n "${PROLE_TEST_CFG:-}" ]]; then
    if [[ "$PROLE_TEST_CFG" == "true" || "$PROLE_TEST_CFG" == "1" ]]; then
      CFG_PATH="$ROOT_DIR/tests/fixtures/test-knoe.cfg"
    else
      CFG_PATH="$PROLE_TEST_CFG"
    fi
  elif [[ -n "${KNOE_CONF:-}" && -f "$KNOE_CONF/knoe.cfg" ]]; then
    CFG_PATH="$KNOE_CONF/knoe.cfg"
  elif [[ -f "$ROOT_DIR/conf/knoe.cfg" ]]; then
    CFG_PATH="$ROOT_DIR/conf/knoe.cfg"
  else
    echo "ERROR: knoe.cfg not found. Pass a path or set KNOE_CONF." >&2
    exit 1
  fi
fi

CFG_DIR="$(cd "$(dirname "$CFG_PATH")" && pwd)"
if [[ -z "${KNOE_CONF:-}" ]]; then
  KNOE_CONF="$CFG_DIR"
  export KNOE_CONF
fi

TMP_CFG=""
if [[ "${PROLE_CFG_INPLACE:-}" != "true" ]]; then
  TMP_CFG="$(mktemp -t knoe.cfg.XXXXXX)"
  cp "$CFG_PATH" "$TMP_CFG"
  CFG_PATH="$TMP_CFG"
  trap '[[ -n "$TMP_CFG" ]] && rm -f "$TMP_CFG"' EXIT
fi

echo "=============================================="
echo "  SILENT INSTALL TEST"
echo "  knoe.cfg: $CFG_PATH"
echo "=============================================="
echo ""

echo "1. Validating knoe.cfg..."
python3 - "$CFG_PATH" <<'PY'
import configparser
import os
import sys
from pathlib import Path

cfg_path = Path(sys.argv[1]).expanduser()
if not cfg_path.exists():
    print(f"ERROR: knoe.cfg not found at {cfg_path}", file=sys.stderr)
    sys.exit(1)

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

required_sections = ["Inputs", "Global"]
missing_sections = [s for s in required_sections if not cfg.has_section(s)]
if missing_sections:
    print(f"ERROR: Missing sections: {', '.join(missing_sections)}", file=sys.stderr)
    sys.exit(1)

required_inputs = [
    "build.deploy_env",
    "build.run_build",
    "dependencies.auto_install_missing",
    "dependencies.brew.install",
    "dependencies.docker.install",
    "dependencies.k3d.install",
    "dependencies.verify_all",
    "disk_selection.disk_type",
    "disk_selection.removable_mount",
    "disk_selection.local_path",
    "env_setup.NAMESPACE",
    "env_setup.KNOE_CONF",
    "env_setup.PROLE_DATA",
    "env_setup.KNOE_HOME",
    "env_setup.PROLE_LOGS",
    "env_setup.KNOE_SERVICE",
    "init_cluster.at_rest_encryption_enabled",
    "init_cluster.cluster_env",
    "init_cluster.kerberos_enabled",
    "init_cluster.start_cluster",
    "init_cluster.supabase_enabled",
    "init_cnpg_deploy.force_rollout",
    "init_cnpg_deploy.run_deploy",
    "init_db_build.run_build",
    "init_password.db_host_port",
    "init_password.db_namespace",
    "init_password.db_password",
    "init_password.db_password_confirm",
    "init_password.db_username",
    "init_password.generate_ssh_key",
    "init_scripts.run_scripts",
    "kerberos_config.enabled",
    "kerberos_config.init_authority",
    "kerberos_config.kdc",
    "kerberos_config.password",
    "kerberos_config.realm",
    "kerberos_config.test_connection",
    "kerberos_config.user",
    "network_scan.run",
]

missing_inputs = [k for k in required_inputs if not cfg.has_option("Inputs", k)]
if missing_inputs:
    print("ERROR: Missing Inputs keys:", file=sys.stderr)
    for key in missing_inputs:
        print(f"  - {key}", file=sys.stderr)
    sys.exit(1)

required_non_empty = {
    "Inputs": [
        "env_setup.KNOE_HOME",
        "env_setup.KNOE_CONF",
        "env_setup.PROLE_DATA",
        "env_setup.PROLE_LOGS",
        "env_setup.KNOE_SERVICE",
        "env_setup.NAMESPACE",
        "init_password.db_namespace",
        "init_password.db_username",
        "init_password.db_host_port",
    ],
    "Global": [
        "KNOE_HOME",
        "KNOE_DB_USER",
        "CLUSTER_ENV",
        "NAMESPACE",
    ],
}

def is_anchor(val: str) -> bool:
    val = val.strip()
    return val.startswith("${OPENBAO:") or val.startswith("${KNOE_SECRET:")

errors = []
warnings = []

for section, keys in required_non_empty.items():
    for key in keys:
        if not cfg.has_option(section, key):
            errors.append(f"{section}.{key} missing")
            continue
        value = cfg.get(section, key, fallback="").strip()
        if not value:
            errors.append(f"{section}.{key} is empty")

secret_keys = {
    ("Inputs", "init_password.db_password"),
    ("Inputs", "init_password.db_password_confirm"),
    ("Global", "DB_PASSWORD"),
}

for section, key in secret_keys:
    if not cfg.has_option(section, key):
        errors.append(f"{section}.{key} missing")
        continue
    value = cfg.get(section, key, fallback="").strip()
    if not value:
        errors.append(f"{section}.{key} is empty")
    elif is_anchor(value):
        warnings.append(f"{section}.{key} is anchored ({value})")

pw = cfg.get("Inputs", "init_password.db_password", fallback="")
pw_confirm = cfg.get("Inputs", "init_password.db_password_confirm", fallback="")
if pw and pw_confirm and pw != pw_confirm:
    errors.append("Inputs.init_password.db_password does not match init_password.db_password_confirm")

if errors:
    print("ERROR: knoe.cfg validation failed:", file=sys.stderr)
    for err in errors:
        print(f"  - {err}", file=sys.stderr)
    sys.exit(1)

if warnings:
    print("WARN: knoe.cfg contains anchored secrets:")
    for warn in warnings:
        print(f"  - {warn}")

print("OK: knoe.cfg validation")
PY

echo ""
echo "2. Running silent install..."
if "$ROOT_DIR/install.sh" -S -c "$CFG_PATH"; then
  echo "OK: Silent install completed"
else
  echo "ERROR: Silent install failed" >&2
  exit 1
fi

echo ""
echo "3. Blessing sanitized gold config..."
python3 - "$CFG_PATH" "$KNOE_CONF/knoe-db/knoe.cfg" <<'PY'
import re
import sys
from pathlib import Path
import configparser

source = Path(sys.argv[1]).expanduser()
target = Path(sys.argv[2]).expanduser()
target.parent.mkdir(parents=True, exist_ok=True)

text = source.read_text()

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

var_pattern = re.compile(r"\$(\w+)|\$\{(\w+)\}")

def expand_cfg_value(val: str, cfg_vars: dict) -> str:
    if val is None:
        return ""
    raw = str(val)
    def repl(match):
        var = match.group(1) or match.group(2)
        return str(cfg_vars.get(var, match.group(0)))
    out = raw
    for _ in range(5):
        new = var_pattern.sub(repl, out)
        if new == out:
            break
        out = new
    return out

cfg_vars = {}
for section in cfg.sections():
    for k, v in cfg.items(section):
        if k in cfg_vars:
            continue
        cfg_vars[k] = expand_cfg_value(v, cfg_vars)

namespace = (
    expand_cfg_value(cfg.get("Global", "NAMESPACE", fallback=""), cfg_vars)
    or expand_cfg_value(cfg.get("Inputs", "env_setup.NAMESPACE", fallback=""), cfg_vars)
    or "default"
).strip() or "default"

def bao(leaf: str, key: str) -> str:
    return f"${{OPENBAO:kv/knoe/{namespace}/{leaf}#{key}}}"

secret_map = {
    "init_password.db_password": bao("db", "password"),
    "init_password.db_password_confirm": bao("db", "password"),
    "kerberos_config.password": bao("kerberos", "password"),
    "DB_PASSWORD": bao("db", "password"),
    "PASSWORD": bao("kerberos", "password"),
    "GRAFANA_ADMIN_PASSWORD": bao("monitoring", "grafana_admin_password"),
}

pattern = re.compile(r"^(?P<lead>\s*)(?P<key>[^=]+?)(?P<pre>\s*)=(?P<post>\s*).*$")
out_lines = []

for line in text.splitlines():
    stripped = line.strip()
    if not stripped or stripped.startswith((';', '#')) or '=' not in line:
        out_lines.append(line)
        continue
    m = pattern.match(line)
    if not m:
        out_lines.append(line)
        continue
    key = m.group("key").strip()
    lead = m.group("lead")
    pre = m.group("pre")
    post = m.group("post")
    if key in secret_map:
        value = secret_map[key]
        out_lines.append(f"{lead}{key}{pre}={post}{value}")
        continue
    if re.search(r"(?i)(?:^|[_.])(password|token|secret)$", key):
        out_lines.append(f"{lead}{key}{pre}={post}${{OPENBAO:REDACTED}}")
        continue
    out_lines.append(line)

target.write_text("\n".join(out_lines) + "\n")
print(f"OK: Wrote sanitized gold config: {target}")
PY

echo ""
echo "=============================================="
echo "  OK: Silent Install Test Completed"
echo "=============================================="
echo ""