prole/mock_val/build-a-bao.sh

#!/usr/bin/env bash

set -euo pipefail

# build-a-bao.sh
# Purpose:
# - Decrypt temporary secrets stored in the active config file
# - Store them in OpenBao for the current namespace
# - Replace config-file secrets with OpenBao placeholders

SCRIPT_DIR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)

# Load environment and config via knoe_cfg.sh
# shellcheck disable=SC1090
source "$SCRIPT_DIR/knoe_cfg.sh"

cfg_path=""
if [[ -n "${KNOE_CONF:-}" ]]; then
  cfg_path="$(_knoe_cfg_select_cfg_file "$KNOE_CONF")"
elif [[ -n "${KNOE_HOME:-}" ]]; then
  cfg_path="$(_knoe_cfg_select_cfg_file "$KNOE_HOME/conf")"
else
  cfg_path="$(_knoe_cfg_select_cfg_file "$SCRIPT_DIR/../conf")"
fi

if [[ -z "$cfg_path" || ! -f "$cfg_path" ]]; then
  echo "ERROR: config file not found. Set KNOE_CONF or KNOE_HOME." >&2
  exit 1
fi

tmp_env="$(mktemp)"
python3 - "$cfg_path" >"$tmp_env" <<'PY'
import base64
import configparser
import getpass
import os
import platform
import re
import shlex
import subprocess
import sys
from pathlib import Path

KNOE_SECRET_PREFIX = "${KNOE_SECRET:"
KNOE_SECRET_SUFFIX = "}"
OPENBAO_PREFIX = "${OPENBAO:"
OPENBAO_SUFFIX = "}"
KNOE_SECRET_VERSION = "v1"
KNOE_SECRET_SERVICE = "knoe-installer"
KNOE_SECRET_KEY_FILE = Path.home() / ".knoe" / "secrets" / "installer.key"

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_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(KNOE_SECRET_KEY_FILE)

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 ""
    try:
        from cryptography.hazmat.primitives.ciphers.aead import AESGCM
        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)
        return aesgcm.decrypt(nonce, ciphertext, None).decode("utf-8")
    except Exception:
        return ""

cfg_path = Path(sys.argv[1]).expanduser()
cfg = configparser.ConfigParser(interpolation=None)
cfg.optionxform = str
cfg.read(cfg_path)

def get_val(section: str, key: str) -> str:
    if cfg.has_option(section, key):
        return cfg.get(section, key, fallback="").strip()
    return ""

namespace = get_val("Global", "NAMESPACE") or get_val("Inputs", "env_setup.NAMESPACE") or get_val("Database Creation", "NAMESPACE") or "default"
db_user = get_val("Inputs", "init_password.db_username") or get_val("Global", "KNOE_DB_USER")
db_pass = get_val("Inputs", "init_password.db_password") or get_val("Global", "DB_PASSWORD")
krb_realm = get_val("Inputs", "kerberos_config.realm") or get_val("Kerberos Authentication", "REALM")
krb_kdc = get_val("Inputs", "kerberos_config.kdc") or get_val("Kerberos Authentication", "KDC")
krb_user = get_val("Inputs", "kerberos_config.user") or get_val("Kerberos Authentication", "USER")
krb_pass = get_val("Inputs", "kerberos_config.password") or get_val("Kerberos Authentication", "PASSWORD")
grafana_pass = get_val("Monitoring", "GRAFANA_ADMIN_PASSWORD")
at_rest = get_val("Inputs", "init_cluster.at_rest_encryption_enabled") or get_val("Optional Features", "AT_REST_ENCRYPTION_ENABLED")

def derive_child_id(ns: str) -> str:
    if not ns:
        return ""
    match = re.match(r"^knoe-db-([A-Za-z0-9]+)$", ns)
    if not match:
        return ""
    return match.group(1).upper()

def derive_child_realm(ns: str, parent_realm: str) -> str:
    if not parent_realm:
        return ""
    child_id = derive_child_id(ns)
    if not child_id:
        return ""
    parent_realm = parent_realm.upper()
    if parent_realm.startswith(child_id + "."):
        return parent_realm
    return f"{child_id}.{parent_realm}"

child_realm = derive_child_realm(namespace, krb_realm)
if child_realm:
    krb_realm = child_realm

if db_pass and not is_openbao_ref(db_pass):
    db_pass = decrypt_knoe_secret(db_pass)
else:
    db_pass = ""
if krb_pass and not is_openbao_ref(krb_pass):
    krb_pass = decrypt_knoe_secret(krb_pass)
else:
    krb_pass = ""
if grafana_pass and not is_openbao_ref(grafana_pass):
    grafana_pass = decrypt_knoe_secret(grafana_pass)
else:
    grafana_pass = ""

def emit(key: str, value: str):
    if value is None:
        value = ""
    print(f"export {key}={shlex.quote(str(value))}")

emit("NAMESPACE", namespace)
emit("KNOE_DB_USER", db_user)
emit("KNOE_DB_PASSWORD", db_pass)
emit("KRB5_REALM", krb_realm)
emit("KRB5_KDC", krb_kdc)
emit("KRB5_ADMIN", krb_kdc)
emit("KRB5_USER", krb_user)
emit("KRB5_PASSWORD", krb_pass)
emit("GRAFANA_ADMIN_PASSWORD", grafana_pass)
emit("AT_REST_ENCRYPTION_ENABLED", at_rest)
PY

# shellcheck disable=SC1090
source "$tmp_env"
rm -f "$tmp_env"

export KNOE_HOME="${KNOE_HOME:-$(cd "$SCRIPT_DIR/.." && pwd)}"
export KNOE_SERVICE="${KNOE_SERVICE:-$KNOE_HOME}"
NAMESPACE="${PROLE_NAMESPACE}"

db_pass="${KNOE_DB_PASSWORD:-}"
if [[ -z "$db_pass" ]]; then
  echo "[WARN] No database password found in knoe.cfg; skipping db secret write."
fi

if [[ -n "$db_pass" ]]; then
  printf '%s\n' "$db_pass" | "$SCRIPT_DIR/init_openbao.sh" initialize
else
  "$SCRIPT_DIR/init_openbao.sh" initialize </dev/null
fi

python3 - "$cfg_path" <<'PY'
import configparser
import re
import sys

cfg_path = sys.argv[1]
cfg = configparser.ConfigParser(interpolation=None)
cfg.optionxform = str
cfg.read(cfg_path)

def get_val(section: str, key: str) -> str:
    if cfg.has_option(section, key):
        return cfg.get(section, key, fallback="").strip()
    return ""

namespace = get_val("Global", "NAMESPACE") or get_val("Inputs", "env_setup.NAMESPACE") or get_val("Database Creation", "NAMESPACE") or "default"
def openbao_placeholder(ns: str, leaf: str, key: str) -> str:
    return f"${{OPENBAO:kv/knoe/{ns}/{leaf}#{key}}}"

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

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

with open(cfg_path, "r", encoding="utf-8") as f:
    for line in f.read().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:
            out_lines.append(f"{lead}{key}{pre}={post}{secret_map[key]}")
            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)

with open(cfg_path, "w", encoding="utf-8") as f:
    f.write("\n".join(out_lines) + "\n")
print(f"[OK] Updated secrets in {cfg_path}")
PY