mirror of
https://github.com/dredx/prole.git
synced 2026-09-23 10:13:58 +00:00
6f99f95f84
Fix _deployment_mode_hint() to correctly map cluster_env=service → k3s mode. Previously 'service' was not handled in the normalized_env checks, causing fallthrough to build.deploy_env='Dev' → mode='dev' → k3d dependency required. On a k3s node (myrddin/merlin/gandalf) k3d is not installed and the DependenciesMilestone fatally aborted the install. Also fix get_required_dependency_ids(): k3s mode does not require k3d (k3s is provisioned on the cluster nodes by Ansible, not by the installer binary). Fixes: install.sh --mode k3s --reset failing with 'Dependencies unresolved after install attempts. Required: k3d' Co-authored-by: Junie <junie@jetbrains.com>
2261 lines
72 KiB
Python
2261 lines
72 KiB
Python
"""
|
||
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 argparse
|
||
import base64
|
||
import curses
|
||
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
|
||
|
||
LOGGER = logging.getLogger(__name__)
|
||
|
||
|
||
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)
|
||
- `img/knoe.png`
|
||
- legacy `ui.icon`
|
||
|
||
Defaults to img/knoe.png under repo root if not set or missing.
|
||
"""
|
||
# Prefer new key `icon`, then fallback to knoe.png, then old `ui.icon`
|
||
rel = get_config_value("icon") or "img/knoe.png"
|
||
p = get_resource_path(rel)
|
||
if p.exists():
|
||
return p
|
||
|
||
# Fallback to legacy ui.icon if set
|
||
legacy_rel = get_config_value("ui.icon")
|
||
if legacy_rel:
|
||
p = get_resource_path(legacy_rel)
|
||
if p.exists():
|
||
return p
|
||
|
||
# Final fallback
|
||
return get_resource_path("img/knoe.png")
|
||
|
||
|
||
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 = get_resource_path(rel)
|
||
if p.exists():
|
||
return p
|
||
# fallback
|
||
return get_resource_path("img/knoeLogoSepia.png")
|
||
|
||
|
||
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:
|
||
import knoe.config as _self
|
||
base_path = _self.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 _preserve_cfg_expr_for_persistence(
|
||
raw_value: str | None,
|
||
expanded_value: str | None = None,
|
||
) -> str:
|
||
"""Prefer raw tokenized value (e.g. `${HOME}`) when persisting config."""
|
||
|
||
raw = "" if raw_value is None else str(raw_value).strip()
|
||
if "$" in raw:
|
||
return raw
|
||
if expanded_value is None:
|
||
return raw
|
||
return str(expanded_value).strip()
|
||
|
||
|
||
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:
|
||
"""Extract an Ansible Vault inline block for *key* from *text*.
|
||
|
||
Returns the raw vault block string (including the ``$ANSIBLE_VAULT``
|
||
header line) when found, or an empty string otherwise.
|
||
"""
|
||
lines = text.splitlines()
|
||
in_block = False
|
||
block_lines: list[str] = []
|
||
for line in lines:
|
||
if not in_block:
|
||
if re.match(rf"^{re.escape(key)}:\s*$", line):
|
||
in_block = True
|
||
else:
|
||
stripped = line.strip()
|
||
if not stripped:
|
||
break
|
||
if not line.startswith(" ") and not line.startswith("\t"):
|
||
break
|
||
block_lines.append(stripped)
|
||
if block_lines and block_lines[0].startswith("$ANSIBLE_VAULT"):
|
||
return "\n".join(block_lines)
|
||
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
|
||
|
||
|
||
def _run_as_root_noninteractive(command: str) -> str:
|
||
return (
|
||
"if command -v sudo >/dev/null 2>&1; then "
|
||
f"sudo -n {command}; "
|
||
f"else {command}; fi"
|
||
)
|
||
|
||
|
||
def _apt_install(pkg: str) -> str:
|
||
return _run_as_root_noninteractive(
|
||
f"env DEBIAN_FRONTEND=noninteractive apt-get install -y {pkg}"
|
||
)
|
||
|
||
|
||
def _run_as_root_noninteractive_shell(command: str) -> str:
|
||
return _run_as_root_noninteractive(f"bash -c {shlex.quote(command)}")
|
||
|
||
|
||
def _with_clear_failure(cmd: str, message: str) -> str:
|
||
return f"{cmd} || {{ echo \"[ERROR] {message}\"; exit 1; }}"
|
||
|
||
|
||
_APT_VENDOR_REPOSITORIES: dict[str, dict[str, str]] = {
|
||
"google-cloud-cli": {
|
||
"name": "Google Cloud SDK",
|
||
"reason": "required for gcloud when package is absent from distro repositories",
|
||
"key_url": "https://packages.cloud.google.com/apt/doc/apt-key.gpg",
|
||
"keyring": "/etc/apt/keyrings/google-cloud.gpg",
|
||
"source_file": "google-cloud-sdk.list",
|
||
"repo_line": "deb [signed-by=/etc/apt/keyrings/google-cloud.gpg] https://packages.cloud.google.com/apt cloud-sdk main",
|
||
},
|
||
"opentofu": {
|
||
"name": "OpenTofu",
|
||
"reason": "required when OpenTofu package is absent from distro repositories",
|
||
"key_url": "https://packages.opentofu.org/opentofu/tofu/gpgkey",
|
||
"keyring": "/etc/apt/keyrings/opentofu.gpg",
|
||
"source_file": "opentofu.list",
|
||
"repo_line": "deb [signed-by=/etc/apt/keyrings/opentofu.gpg] https://packages.opentofu.org/opentofu/tofu/any/ any main",
|
||
},
|
||
}
|
||
|
||
|
||
def _apt_install_with_vendor_repo_fallback(pkg: str) -> str:
|
||
vendor = _APT_VENDOR_REPOSITORIES.get(pkg)
|
||
if not vendor:
|
||
return _apt_install(pkg)
|
||
|
||
keyring = vendor["keyring"]
|
||
source_file = f"/etc/apt/sources.list.d/{vendor['source_file']}"
|
||
repo_line = vendor["repo_line"]
|
||
key_url = vendor["key_url"]
|
||
|
||
return "\n".join(
|
||
[
|
||
f"if apt-cache show {shlex.quote(pkg)} >/dev/null 2>&1; then",
|
||
f" echo \"[INFO] apt package '{pkg}' found in configured apt metadata; installing directly.\"",
|
||
f" {_with_clear_failure(_apt_install(pkg), f'Failed to install apt package {pkg}.')}",
|
||
"else",
|
||
(
|
||
" echo \"[INFO] apt package "
|
||
f"'{pkg}' missing from configured apt metadata; adding vendor repository "
|
||
f"'{vendor['name']}' ({vendor['reason']}).\""
|
||
),
|
||
(
|
||
" "
|
||
+ _with_clear_failure(
|
||
_run_as_root_noninteractive("install -m 0755 -d /etc/apt/keyrings"),
|
||
"Failed to create /etc/apt/keyrings.",
|
||
)
|
||
),
|
||
(
|
||
" "
|
||
+ _with_clear_failure(
|
||
_run_as_root_noninteractive_shell(
|
||
f"curl -fsSL {shlex.quote(key_url)} | gpg --dearmor -o {shlex.quote(keyring)}"
|
||
),
|
||
f"Failed to install apt keyring for vendor repository {vendor['name']}.",
|
||
)
|
||
),
|
||
(
|
||
" "
|
||
+ _with_clear_failure(
|
||
_run_as_root_noninteractive(f"chmod a+r {shlex.quote(keyring)}"),
|
||
f"Failed to set permissions on keyring {keyring}.",
|
||
)
|
||
),
|
||
(
|
||
" "
|
||
+ _with_clear_failure(
|
||
_run_as_root_noninteractive_shell(
|
||
f"printf '%s\\n' {shlex.quote(repo_line)} > {shlex.quote(source_file)}"
|
||
),
|
||
f"Failed to configure apt source list {source_file}.",
|
||
)
|
||
),
|
||
(
|
||
" echo \"[INFO] Added vendor apt repository "
|
||
f"'{vendor['name']}': {repo_line}\""
|
||
),
|
||
(
|
||
" "
|
||
+ _with_clear_failure(
|
||
_run_as_root_noninteractive("apt-get update"),
|
||
f"Failed to refresh apt metadata after adding vendor repository {vendor['name']}.",
|
||
)
|
||
),
|
||
(
|
||
" "
|
||
+ _with_clear_failure(
|
||
_apt_install(pkg),
|
||
f"Failed to install apt package {pkg} after repository bootstrap.",
|
||
)
|
||
),
|
||
"fi",
|
||
]
|
||
)
|
||
|
||
|
||
def _rpm_install(manager: str, pkg: str) -> str:
|
||
return _run_as_root_noninteractive(f"{manager} install -y {pkg}")
|
||
|
||
|
||
_MACOS_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": "op",
|
||
"name": "1password-cli",
|
||
"parent": "brew",
|
||
"description": "1Password CLI for secret management",
|
||
"url": "https://developer.1password.com/docs/cli",
|
||
"install_cmd": "brew install 1password-cli",
|
||
"check_cmd": "op --version",
|
||
"bin": "op",
|
||
},
|
||
{
|
||
"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": "gcloud",
|
||
"name": "gcloud",
|
||
"parent": "brew",
|
||
"description": "Google Cloud SDK CLI for GKE authentication and context setup",
|
||
"url": "https://cloud.google.com/sdk/docs/install",
|
||
"install_cmd": "brew install --cask google-cloud-sdk || brew install google-cloud-sdk",
|
||
"check_cmd": "gcloud --version",
|
||
"bin": "gcloud",
|
||
},
|
||
{
|
||
"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",
|
||
},
|
||
{
|
||
"id": "containerd",
|
||
"name": "containerd",
|
||
"parent": "brew",
|
||
"description": "Industry-standard container runtime",
|
||
"url": "https://containerd.io",
|
||
"install_cmd": "brew install containerd",
|
||
"check_cmd": "containerd --version",
|
||
"bin": "containerd",
|
||
},
|
||
{
|
||
"id": "docker-buildx",
|
||
"name": "docker-buildx",
|
||
"parent": "brew",
|
||
"description": "Docker CLI plugin for extended build capabilities",
|
||
"url": "https://github.com/docker/buildx",
|
||
"install_cmd": "brew install docker-buildx",
|
||
"check_cmd": "docker buildx version",
|
||
"bin": "docker-buildx",
|
||
},
|
||
]
|
||
|
||
|
||
def _linux_dependencies(manager: str) -> list[dict]:
|
||
if manager not in {"apt", "dnf", "yum"}:
|
||
return []
|
||
|
||
install = (
|
||
_apt_install_with_vendor_repo_fallback
|
||
if manager == "apt"
|
||
else lambda pkg: _rpm_install(manager, pkg)
|
||
)
|
||
deps = [
|
||
{
|
||
"id": "python",
|
||
"name": "python",
|
||
"description": "Python programming language",
|
||
"url": "https://www.python.org",
|
||
"install_cmd": install("python3"),
|
||
"check_cmd": "python3 --version",
|
||
"bin": "python3",
|
||
},
|
||
{
|
||
"id": "op",
|
||
"name": "1password-cli",
|
||
"description": "1Password CLI for secret management",
|
||
"url": "https://developer.1password.com/docs/cli",
|
||
"install_cmd": install("1password-cli"),
|
||
"check_cmd": "op --version",
|
||
"bin": "op",
|
||
},
|
||
{
|
||
"id": "kubectl",
|
||
"name": "kubectl",
|
||
"description": "Kubernetes command-line tool",
|
||
"url": "https://kubernetes.io/docs/reference/kubectl/",
|
||
"install_cmd": install("kubectl"),
|
||
"check_cmd": "kubectl version --client",
|
||
"bin": "kubectl",
|
||
},
|
||
{
|
||
"id": "kubectx",
|
||
"name": "kubectx",
|
||
"description": "Tool to switch between Kubernetes contexts",
|
||
"url": "https://github.com/ahmetb/kubectx",
|
||
"install_cmd": install("kubectx"),
|
||
"check_cmd": "kubectx -h",
|
||
"bin": "kubectx",
|
||
},
|
||
{
|
||
"id": "gcloud",
|
||
"name": "gcloud",
|
||
"description": "Google Cloud SDK CLI for GKE authentication and context setup",
|
||
"url": "https://cloud.google.com/sdk/docs/install",
|
||
"install_cmd": install("google-cloud-cli"),
|
||
"check_cmd": "gcloud --version",
|
||
"bin": "gcloud",
|
||
},
|
||
{
|
||
"id": "docker",
|
||
"name": "Docker",
|
||
"description": "Container platform for running Knoe services",
|
||
"url": "https://www.docker.com",
|
||
"install_cmd": install("docker.io" if manager == "apt" else "docker"),
|
||
"check_cmd": "docker --version",
|
||
"bin": "docker",
|
||
},
|
||
{
|
||
"id": "k3d",
|
||
"name": "k3d",
|
||
"description": "Lightweight wrapper to run k3s in Docker",
|
||
"url": "https://k3d.io",
|
||
"install_cmd": install("k3d"),
|
||
"check_cmd": "k3d --version",
|
||
"bin": "k3d",
|
||
},
|
||
{
|
||
"id": "opentofu",
|
||
"name": "OpenTofu",
|
||
"description": "Open-source infrastructure as code tool",
|
||
"url": "https://opentofu.org",
|
||
"install_cmd": install("opentofu"),
|
||
"check_cmd": "tofu --version",
|
||
"bin": "tofu",
|
||
},
|
||
]
|
||
|
||
if manager == "apt":
|
||
deps.append(
|
||
{
|
||
"id": "gke-gcloud-auth-plugin",
|
||
"name": "gke-gcloud-auth-plugin",
|
||
"description": "GKE auth plugin required by kubectl for gcloud-based authentication",
|
||
"url": "https://cloud.google.com/kubernetes-engine/docs/how-to/cluster-access-for-kubectl",
|
||
"install_cmd": install("google-cloud-cli-gke-gcloud-auth-plugin"),
|
||
"check_cmd": "gke-gcloud-auth-plugin --version",
|
||
"bin": "gke-gcloud-auth-plugin",
|
||
}
|
||
)
|
||
|
||
return deps
|
||
|
||
|
||
_DEP_PLATFORM_CACHE: Optional[dict] = None
|
||
|
||
|
||
def _detect_linux_package_manager() -> tuple[Optional[str], Optional[str]]:
|
||
if shutil.which("apt-get") and shutil.which("dpkg"):
|
||
return "apt", "apt/deb"
|
||
if shutil.which("dnf"):
|
||
return "dnf", "rpm"
|
||
if shutil.which("yum"):
|
||
return "yum", "rpm"
|
||
if shutil.which("rpm"):
|
||
return "rpm", "rpm"
|
||
return None, None
|
||
|
||
|
||
def detect_dependency_platform(refresh: bool = False) -> dict:
|
||
global _DEP_PLATFORM_CACHE
|
||
if _DEP_PLATFORM_CACHE is not None and not refresh:
|
||
return dict(_DEP_PLATFORM_CACHE)
|
||
|
||
os_raw = (platform.system() or "").lower()
|
||
arch = (platform.machine() or "unknown").lower()
|
||
info: dict[str, Optional[str] | bool] = {
|
||
"os": None,
|
||
"arch": arch,
|
||
"package_manager": None,
|
||
"package_family": None,
|
||
"supported": False,
|
||
"reason": None,
|
||
}
|
||
|
||
if os_raw == "darwin":
|
||
info.update(
|
||
{
|
||
"os": "macos",
|
||
"package_manager": "brew",
|
||
"package_family": "brew",
|
||
"supported": True,
|
||
}
|
||
)
|
||
elif os_raw == "linux":
|
||
manager, family = _detect_linux_package_manager()
|
||
info.update({"os": "linux", "package_manager": manager, "package_family": family})
|
||
if manager in {"apt", "dnf", "yum"}:
|
||
info["supported"] = True
|
||
elif manager == "rpm":
|
||
info["reason"] = (
|
||
"Linux RPM database detected but no supported package installer "
|
||
"(dnf/yum) was found."
|
||
)
|
||
elif shutil.which("brew"):
|
||
if _parse_bool(os.environ.get("PROLE_LINUX_ALLOW_BREW"), False):
|
||
info.update(
|
||
{
|
||
"package_manager": "brew",
|
||
"package_family": "brew",
|
||
"supported": True,
|
||
"reason": "Linux Homebrew explicitly enabled via PROLE_LINUX_ALLOW_BREW.",
|
||
}
|
||
)
|
||
else:
|
||
info["reason"] = (
|
||
"Linux Homebrew detected but native package managers (apt/dnf/yum) "
|
||
"were not found. Refusing brew fallback by default."
|
||
)
|
||
else:
|
||
info["reason"] = "No supported Linux package manager detected (apt/dnf/yum)."
|
||
else:
|
||
info["reason"] = f"Unsupported operating system: {platform.system() or 'unknown'}"
|
||
|
||
_DEP_PLATFORM_CACHE = dict(info)
|
||
return dict(info)
|
||
|
||
|
||
def get_dependency_milestone_title() -> str:
|
||
info = detect_dependency_platform()
|
||
if info.get("os") == "linux":
|
||
return "Linux Dependencies"
|
||
if info.get("os") == "macos":
|
||
return "Mac OS Dependencies"
|
||
return "Dependencies"
|
||
|
||
|
||
def get_dependency_platform_summary() -> str:
|
||
info = detect_dependency_platform()
|
||
return (
|
||
f"os={info.get('os') or 'unknown'}, "
|
||
f"arch={info.get('arch') or 'unknown'}, "
|
||
f"package_family={info.get('package_family') or 'unknown'}, "
|
||
f"package_manager={info.get('package_manager') or 'unknown'}"
|
||
)
|
||
|
||
|
||
def get_platform_dependencies() -> list[dict]:
|
||
info = detect_dependency_platform()
|
||
if not info.get("supported"):
|
||
return []
|
||
|
||
if info.get("os") == "macos":
|
||
return [dict(dep) for dep in _MACOS_DEPENDENCIES]
|
||
|
||
if info.get("os") == "linux":
|
||
manager = info.get("package_manager")
|
||
if manager == "brew":
|
||
return [dict(dep) for dep in _MACOS_DEPENDENCIES if dep["id"] != "brew"]
|
||
return _linux_dependencies(str(manager))
|
||
|
||
return []
|
||
|
||
|
||
def _deployment_mode_hint(inputs: dict | None = None) -> str:
|
||
from knoe.core.env import _normalize_cluster_env
|
||
|
||
data = inputs or {}
|
||
cluster_env_raw = str(data.get("init_cluster.cluster_env", "")).strip()
|
||
normalized_env = (_normalize_cluster_env(cluster_env_raw) or "").strip().lower()
|
||
|
||
if normalized_env in {"prod", "production", "k8s"}:
|
||
return "gke"
|
||
if normalized_env in {"dev", "k3d"}:
|
||
return normalized_env
|
||
if normalized_env == "service":
|
||
return "k3s"
|
||
if normalized_env == "k3s":
|
||
return "k3s"
|
||
|
||
deployment_mode = str(
|
||
data.get("build.deploy_env")
|
||
or data.get("deployment_mode")
|
||
or data.get("DEPLOYMENT_MODE")
|
||
or ""
|
||
).strip().lower()
|
||
if deployment_mode in {"prod", "production", "k8s", "gke"}:
|
||
return "gke"
|
||
if deployment_mode in {"dev", "k3d", "k3s"}:
|
||
return deployment_mode
|
||
|
||
return "generic"
|
||
|
||
|
||
def get_required_dependency_ids(inputs: dict | None = None) -> set[str]:
|
||
mode = _deployment_mode_hint(inputs)
|
||
|
||
if mode == "min":
|
||
return {"python", "containerd", "docker-buildx"}
|
||
|
||
base_required = {"python", "op", "kubectl", "kubectx", "docker"}
|
||
if mode == "gke":
|
||
required = base_required | {"gcloud"}
|
||
platform_info = detect_dependency_platform()
|
||
if (
|
||
platform_info.get("os") == "linux"
|
||
and platform_info.get("package_manager") == "apt"
|
||
):
|
||
required.add("gke-gcloud-auth-plugin")
|
||
return required
|
||
if mode in {"dev", "k3d"}:
|
||
return base_required | {"k3d"}
|
||
if mode == "k3s":
|
||
return base_required
|
||
return base_required
|
||
|
||
|
||
def get_required_dependencies(inputs: dict | None = None) -> list[dict]:
|
||
dependencies = get_platform_dependencies()
|
||
required_ids = get_required_dependency_ids(inputs)
|
||
return [dep for dep in dependencies if str(dep.get("id")) in required_ids]
|
||
|
||
|
||
def _validate_dependency_backend(dependencies: list[dict] | None = None) -> tuple[bool, str]:
|
||
info = detect_dependency_platform()
|
||
if not info.get("supported"):
|
||
reason = info.get("reason") or "Unsupported dependency backend"
|
||
return False, str(reason)
|
||
|
||
if info.get("os") == "linux":
|
||
deps = dependencies if dependencies is not None else get_platform_dependencies()
|
||
brew_refs = [dep.get("id") for dep in deps if "brew" in str(dep.get("install_cmd") or "")]
|
||
if brew_refs and info.get("package_manager") != "brew":
|
||
return (
|
||
False,
|
||
"Linux dependency set contains Homebrew commands while native package manager "
|
||
f"{info.get('package_manager')} is active: {', '.join([str(r) for r in brew_refs])}",
|
||
)
|
||
return True, ""
|
||
|
||
|
||
DEPENDENCIES = get_platform_dependencies()
|
||
|
||
|
||
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:
|
||
sdk_bin_dir = str(prefix / "share" / "google-cloud-sdk" / "bin")
|
||
if os.path.isdir(sdk_bin_dir) and sdk_bin_dir not in parts:
|
||
prepend.append(sdk_bin_dir)
|
||
except Exception:
|
||
# SDK path is optional
|
||
pass
|
||
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 _augment_env_for_dependency_backend(env: dict | None = None) -> dict:
|
||
info = detect_dependency_platform()
|
||
if info.get("package_manager") == "brew":
|
||
return _augment_env_for_brew(env)
|
||
return env or os.environ.copy()
|
||
|
||
|
||
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_dependency_backend(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_source = DEPENDENCIES if DEPENDENCIES else get_platform_dependencies()
|
||
deps: list[dict] = [dict(d) for d in deps_source]
|
||
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
|
||
|
||
|
||
def _parse_gcp_cfg(path) -> dict:
|
||
"""Parse flat tfvars-style file written by etc/config.py --provider gcp.
|
||
Handles: key = "value" and key = value
|
||
Skips lines starting with # or ;
|
||
Returns empty dict if file missing or unreadable.
|
||
"""
|
||
result = {}
|
||
try:
|
||
for raw in path.read_text().splitlines():
|
||
line = raw.strip()
|
||
if not line or line.startswith("#") or line.startswith(";"):
|
||
continue
|
||
if "=" not in line:
|
||
continue
|
||
k, _, v = line.partition("=")
|
||
key = k.strip()
|
||
val = v.strip().strip('"').strip("'")
|
||
if key:
|
||
result[key] = val
|
||
except Exception:
|
||
pass
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# GCP ncurses TUI — cluster configuration generator
|
||
#
|
||
# Usage (direct):
|
||
# python3 -m knoe.config --mode k8s --provider gcp
|
||
#
|
||
# Usage (via launcher):
|
||
# ./config.sh --mode k8s --provider gcp
|
||
#
|
||
# Flow:
|
||
# 1. Check / perform gcloud auth (device-code, no browser required)
|
||
# 2. Select GCP organization
|
||
# 3. Select project (filtered to org)
|
||
# 4. Select billing account
|
||
# 5. Confirm selection and write flat tfvars-compatible output file
|
||
#
|
||
# Output format:
|
||
# org_id = "<value>"
|
||
# billing_account = "<value>"
|
||
# billing_project = "<value>"
|
||
# project_id = "<value>"
|
||
# ---------------------------------------------------------------------------
|
||
|
||
DEFAULT_GCP_OUTPUT = PROJECT_ROOT / "conf" / "prod" / "gcp.cfg"
|
||
SUPPORTED_MODES = ("k8s", "prod")
|
||
SUPPORTED_PROVIDERS = ("gcp",)
|
||
|
||
|
||
# -- gcloud helpers ----------------------------------------------------------
|
||
|
||
def _gcloud(*args, token: str | None = None) -> tuple[int, list | dict]:
|
||
cmd = ["gcloud", *args, "--format=json", "--quiet"]
|
||
env = os.environ.copy()
|
||
if token:
|
||
env["CLOUDSDK_AUTH_ACCESS_TOKEN"] = token
|
||
try:
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, env=env)
|
||
except FileNotFoundError:
|
||
return 127, []
|
||
except subprocess.TimeoutExpired:
|
||
return 1, []
|
||
if result.returncode != 0:
|
||
return result.returncode, []
|
||
try:
|
||
return 0, json.loads(result.stdout.strip() or "[]")
|
||
except Exception:
|
||
return 1, []
|
||
|
||
|
||
def _gcloud_plain(*args, token: str | None = None) -> tuple[int, str]:
|
||
cmd = ["gcloud", *args, "--quiet"]
|
||
env = os.environ.copy()
|
||
if token:
|
||
env["CLOUDSDK_AUTH_ACCESS_TOKEN"] = token
|
||
try:
|
||
result = subprocess.run(cmd, capture_output=True, text=True, timeout=30, env=env)
|
||
return result.returncode, result.stdout.strip()
|
||
except FileNotFoundError:
|
||
return 127, ""
|
||
except subprocess.TimeoutExpired:
|
||
return 1, ""
|
||
|
||
|
||
def gcloud_available() -> bool:
|
||
rc, _ = _gcloud_plain("version")
|
||
return rc != 127
|
||
|
||
|
||
def active_gcp_account(token: str | None = None) -> str | None:
|
||
rc, data = _gcloud("auth", "list", "--filter=status=ACTIVE", token=token)
|
||
if rc == 0 and data:
|
||
return data[0].get("account", "")
|
||
# With an access token, gcloud auth list may return nothing — verify via token info
|
||
if token:
|
||
return _resolve_token_email(token)
|
||
return None
|
||
|
||
|
||
def _resolve_token_email(token: str) -> str | None:
|
||
"""Resolve the Google account email for an access token via the tokeninfo endpoint."""
|
||
try:
|
||
import urllib.parse
|
||
url = "https://www.googleapis.com/oauth2/v3/tokeninfo"
|
||
req = urllib.request.Request(f"{url}?access_token={urllib.parse.quote(token)}")
|
||
with urllib.request.urlopen(req, timeout=5) as resp:
|
||
data = json.loads(resp.read().decode())
|
||
return data.get("email") or data.get("sub") or "authenticated"
|
||
except Exception:
|
||
return "authenticated"
|
||
|
||
|
||
def _gcloud_print_access_token() -> str | None:
|
||
"""Return the current active gcloud access token via `gcloud auth print-access-token`.
|
||
Returns the token string on success, or None if not authenticated / gcloud unavailable.
|
||
"""
|
||
rc, token = _gcloud_plain("auth", "print-access-token")
|
||
if rc == 0 and token:
|
||
return token
|
||
return None
|
||
|
||
|
||
def fetch_gcp_orgs(token: str | None = None) -> list[dict]:
|
||
rc, data = _gcloud("organizations", "list", token=token)
|
||
if rc != 0 or not data:
|
||
return []
|
||
rows = []
|
||
for d in data:
|
||
org_id = d.get("name", "").replace("organizations/", "")
|
||
name = d.get("displayName", "")
|
||
rows.append({
|
||
"display": f"{name:<40} {org_id}",
|
||
"org_id": org_id,
|
||
"org_name": name,
|
||
})
|
||
return rows
|
||
|
||
|
||
def fetch_gcp_projects(org_id: str | None, token: str | None = None) -> list[dict]:
|
||
args = ["projects", "list"]
|
||
if org_id:
|
||
args += [f"--filter=parent.id={org_id} AND parent.type=organization"]
|
||
rc, data = _gcloud(*args, token=token)
|
||
if rc != 0 or not data:
|
||
return []
|
||
rows = []
|
||
for d in data:
|
||
pid = d.get("projectId", "")
|
||
name = d.get("name", "")
|
||
number = d.get("projectNumber", "")
|
||
rows.append({
|
||
"display": f"{name:<35} {pid:<30} #{number}",
|
||
"project_id": pid,
|
||
"project_name": name,
|
||
"project_number": number,
|
||
})
|
||
rows.sort(key=lambda r: r["project_name"].lower())
|
||
return rows
|
||
|
||
|
||
def fetch_gcp_billing_accounts(token: str | None = None) -> list[dict]:
|
||
rc, data = _gcloud("beta", "billing", "accounts", "list", "--filter=open=true", token=token)
|
||
if rc != 0 or not data:
|
||
rc, data = _gcloud("billing", "accounts", "list", "--filter=open=true", token=token)
|
||
if rc != 0 or not data:
|
||
return []
|
||
rows = []
|
||
for d in data:
|
||
acct_id = d.get("name", "").replace("billingAccounts/", "")
|
||
name = d.get("displayName", "")
|
||
rows.append({
|
||
"display": f"{name:<45} {acct_id}",
|
||
"billing_account_id": acct_id,
|
||
"billing_name": name,
|
||
})
|
||
return rows
|
||
|
||
|
||
# -- ncurses UI primitives ---------------------------------------------------
|
||
|
||
def _tui_draw_header(win, title: str, subtitle: str = ""):
|
||
h, w = win.getmaxyx()
|
||
win.attron(curses.A_BOLD)
|
||
win.addstr(0, 0, title[:w - 1])
|
||
win.attroff(curses.A_BOLD)
|
||
win.addstr(1, 0, ("─" * (w - 1))[:w - 1])
|
||
if subtitle:
|
||
win.addstr(2, 2, subtitle[:w - 3], curses.A_DIM)
|
||
|
||
|
||
def _tui_draw_footer(win, text: str):
|
||
h, w = win.getmaxyx()
|
||
win.addstr(h - 1, 0, text[:w - 1], curses.A_DIM)
|
||
|
||
|
||
def tui_message(win, lines: list[str], wait: bool = True):
|
||
win.erase()
|
||
h, w = win.getmaxyx()
|
||
for i, line in enumerate(lines):
|
||
if i >= h - 2:
|
||
break
|
||
win.addstr(i, 0, line[:w - 1])
|
||
if wait:
|
||
_tui_draw_footer(win, "Press any key to continue…")
|
||
win.refresh()
|
||
win.getch()
|
||
else:
|
||
win.refresh()
|
||
|
||
|
||
def tui_list(win, title: str, items: list[dict], subtitle: str = "") -> int | None:
|
||
"""Scrollable list. Returns selected index or None on quit/cancel."""
|
||
curses.curs_set(0)
|
||
idx = 0
|
||
offset = 0
|
||
header_rows = 3 if subtitle else 2
|
||
|
||
while True:
|
||
win.erase()
|
||
h, w = win.getmaxyx()
|
||
_tui_draw_header(win, title, subtitle)
|
||
list_h = h - header_rows - 1
|
||
|
||
if not items:
|
||
win.addstr(header_rows + 1, 2, "(no items)")
|
||
_tui_draw_footer(win, "q quit")
|
||
win.refresh()
|
||
k = win.getch()
|
||
if k in (ord("q"), ord("Q"), 27):
|
||
return None
|
||
continue
|
||
|
||
if idx < offset:
|
||
offset = idx
|
||
elif idx >= offset + list_h:
|
||
offset = idx - list_h + 1
|
||
|
||
for i, item in enumerate(items[offset: offset + list_h]):
|
||
row = header_rows + i
|
||
abs_i = i + offset
|
||
label = item["display"][:w - 4]
|
||
if abs_i == idx:
|
||
win.attron(curses.A_REVERSE)
|
||
win.addstr(row, 2, f" {label} ")
|
||
win.attroff(curses.A_REVERSE)
|
||
else:
|
||
win.addstr(row, 2, label)
|
||
|
||
scroll_info = f" {idx + 1}/{len(items)}"
|
||
_tui_draw_footer(win, f"↑↓/jk navigate Enter select q quit{scroll_info}")
|
||
win.refresh()
|
||
|
||
k = win.getch()
|
||
if k in (curses.KEY_UP, ord("k")) and idx > 0:
|
||
idx -= 1
|
||
elif k in (curses.KEY_DOWN, ord("j")) and idx < len(items) - 1:
|
||
idx += 1
|
||
elif k == curses.KEY_PPAGE:
|
||
idx = max(0, idx - list_h)
|
||
elif k == curses.KEY_NPAGE:
|
||
idx = min(len(items) - 1, idx + list_h)
|
||
elif k in (curses.KEY_ENTER, 10, 13):
|
||
return idx
|
||
elif k in (ord("q"), ord("Q"), 27):
|
||
return None
|
||
|
||
|
||
def tui_confirm(win, summary: dict, output_path: str, mode: str, provider: str) -> str | None:
|
||
"""Show selection summary, allow editing of the output path, then confirm.
|
||
Returns the final path string or None to cancel.
|
||
"""
|
||
curses.curs_set(1)
|
||
path_buf = list(output_path)
|
||
cursor = len(path_buf)
|
||
|
||
while True:
|
||
win.erase()
|
||
h, w = win.getmaxyx()
|
||
|
||
win.attron(curses.A_BOLD)
|
||
win.addstr(0, 0, f"Config: --mode {mode} --provider {provider} — Confirm & Save"[:w - 1])
|
||
win.attroff(curses.A_BOLD)
|
||
win.addstr(1, 0, ("─" * (w - 1))[:w - 1])
|
||
|
||
rows = [
|
||
("Org ID", summary.get("org_id", "(none)")),
|
||
("Org Name", summary.get("org_name", "")),
|
||
("Project ID", summary.get("project_id", "(none)")),
|
||
("Project Name", summary.get("project_name", "")),
|
||
("Project Number", summary.get("project_number", "")),
|
||
("Billing Account", summary.get("billing_account_id", "(none)")),
|
||
("Billing Name", summary.get("billing_name", "")),
|
||
]
|
||
for i, (label, value) in enumerate(rows):
|
||
row = i + 2
|
||
if row >= h - 4:
|
||
break
|
||
win.addstr(row, 2, f"{label + ':':<18} {value}"[:w - 3])
|
||
|
||
path_row = 2 + len(rows) + 1
|
||
if path_row < h - 2:
|
||
win.addstr(path_row, 2, "Output file: "[:w - 3], curses.A_BOLD)
|
||
path_str = "".join(path_buf)
|
||
win.addstr(path_row, 20, path_str[:w - 22])
|
||
win.move(path_row, 20 + min(cursor, w - 22))
|
||
|
||
_tui_draw_footer(win, "Edit path above Enter to write Esc to cancel")
|
||
win.refresh()
|
||
|
||
k = win.getch()
|
||
if k in (curses.KEY_ENTER, 10, 13):
|
||
curses.curs_set(0)
|
||
return "".join(path_buf)
|
||
elif k == 27:
|
||
curses.curs_set(0)
|
||
return None
|
||
elif k in (curses.KEY_BACKSPACE, 127, 8) and cursor > 0:
|
||
path_buf.pop(cursor - 1)
|
||
cursor -= 1
|
||
elif k == curses.KEY_DC and cursor < len(path_buf):
|
||
path_buf.pop(cursor)
|
||
elif k == curses.KEY_LEFT and cursor > 0:
|
||
cursor -= 1
|
||
elif k == curses.KEY_RIGHT and cursor < len(path_buf):
|
||
cursor += 1
|
||
elif k == curses.KEY_HOME:
|
||
cursor = 0
|
||
elif k == curses.KEY_END:
|
||
cursor = len(path_buf)
|
||
elif 32 <= k < 127:
|
||
path_buf.insert(cursor, chr(k))
|
||
cursor += 1
|
||
|
||
|
||
# -- output writer -----------------------------------------------------------
|
||
|
||
def write_gcp_config(path: str, summary: dict, mode: str, provider: str):
|
||
org_id = summary.get("org_id", "")
|
||
billing_account = summary.get("billing_account_id", "")
|
||
project_id = summary.get("project_id", "")
|
||
org_name = summary.get("org_name", "")
|
||
project_name = summary.get("project_name", "")
|
||
billing_name = summary.get("billing_name", "")
|
||
|
||
lines = [
|
||
f"# Generated by knoe/config.py --mode {mode} --provider {provider}",
|
||
"# Copy org_id, billing_account, billing_project into:",
|
||
"# deploy/gcp/terraform/cloud-setup.auto.tfvars",
|
||
"# Loaded automatically by the installer into conf/gke.cfg [GCP]",
|
||
"",
|
||
f"# Org: {org_name}",
|
||
f'org_id = "{org_id}"',
|
||
"",
|
||
f"# Billing account: {billing_name}",
|
||
f'billing_account = "{billing_account}"',
|
||
"",
|
||
"# Billing project (used for API quota / billing attribution)",
|
||
f'billing_project = "{project_id}"',
|
||
"",
|
||
f"# Selected project: {project_name}",
|
||
f'project_id = "{project_id}"',
|
||
]
|
||
|
||
out = Path(path)
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
out.write_text("\n".join(lines) + "\n")
|
||
|
||
|
||
def tui_credentials(win) -> tuple[str, str] | None:
|
||
"""Two-field credential form: email (visible) + access token (masked).
|
||
Returns (email, token) or None if cancelled.
|
||
Tab/Enter advances between fields; Enter on token field submits.
|
||
"""
|
||
curses.curs_set(1)
|
||
fields = [list(""), list("")] # [email_buf, token_buf]
|
||
cursors = [0, 0]
|
||
active = 0 # which field has focus
|
||
|
||
while True:
|
||
win.erase()
|
||
h, w = win.getmaxyx()
|
||
|
||
win.attron(curses.A_BOLD)
|
||
win.addstr(0, 0, "GCP Authentication"[:w - 1])
|
||
win.attroff(curses.A_BOLD)
|
||
win.addstr(1, 0, ("─" * (w - 1))[:w - 1])
|
||
win.addstr(3, 2, "Enter your Google account email and a valid access token."[:w - 3], curses.A_DIM)
|
||
win.addstr(4, 2, "Tab to move between fields. Enter on the token field to continue."[:w - 3], curses.A_DIM)
|
||
|
||
labels = ["Email: ", "Access token:"]
|
||
row_offsets = [6, 8]
|
||
for i, (label, row) in enumerate(zip(labels, row_offsets)):
|
||
if row >= h - 2:
|
||
break
|
||
attr = curses.A_BOLD if i == active else curses.A_NORMAL
|
||
win.addstr(row, 2, label, attr)
|
||
col = 2 + len(label) + 1
|
||
avail = w - col - 2
|
||
if i == 1:
|
||
display = "*" * len(fields[i])
|
||
else:
|
||
display = "".join(fields[i])
|
||
win.addstr(row, col, display[:avail])
|
||
if i == active:
|
||
cx = min(cursors[i], avail - 1) if avail > 0 else 0
|
||
win.move(row, col + cx)
|
||
|
||
_tui_draw_footer(win, "Tab next field Enter submit Esc cancel")
|
||
win.refresh()
|
||
|
||
k = win.getch()
|
||
if k == 27:
|
||
curses.curs_set(0)
|
||
return None
|
||
elif k == ord("\t"):
|
||
active = 1 - active
|
||
elif k in (curses.KEY_ENTER, 10, 13):
|
||
if active == 0:
|
||
active = 1
|
||
else:
|
||
curses.curs_set(0)
|
||
email = "".join(fields[0]).strip()
|
||
token = "".join(fields[1]).strip()
|
||
if email and token:
|
||
return email, token
|
||
elif k in (curses.KEY_BACKSPACE, 127, 8) and cursors[active] > 0:
|
||
fields[active].pop(cursors[active] - 1)
|
||
cursors[active] -= 1
|
||
elif k == curses.KEY_DC and cursors[active] < len(fields[active]):
|
||
fields[active].pop(cursors[active])
|
||
elif k == curses.KEY_LEFT and cursors[active] > 0:
|
||
cursors[active] -= 1
|
||
elif k == curses.KEY_RIGHT and cursors[active] < len(fields[active]):
|
||
cursors[active] += 1
|
||
elif k == curses.KEY_HOME:
|
||
cursors[active] = 0
|
||
elif k == curses.KEY_END:
|
||
cursors[active] = len(fields[active])
|
||
elif 32 <= k < 127:
|
||
fields[active].insert(cursors[active], chr(k))
|
||
cursors[active] += 1
|
||
|
||
|
||
# -- TUI orchestrator --------------------------------------------------------
|
||
|
||
def _run_gcp_tui(stdscr, output_path: str, mode: str, provider: str):
|
||
curses.start_color()
|
||
curses.use_default_colors()
|
||
stdscr.keypad(True)
|
||
|
||
if not gcloud_available():
|
||
tui_message(stdscr, [
|
||
"ERROR: gcloud CLI not found in PATH.",
|
||
"",
|
||
"Install the Google Cloud SDK and re-run:",
|
||
" ./config.sh --mode k8s --provider gcp",
|
||
])
|
||
return
|
||
|
||
# Try to reuse the current gcloud session via print-access-token first.
|
||
# Falls back to manual credential entry if no active session exists.
|
||
auto_token = _gcloud_print_access_token()
|
||
if auto_token:
|
||
account = active_gcp_account(token=auto_token) or "authenticated"
|
||
token = auto_token
|
||
tui_message(stdscr, [
|
||
f"Authenticated as: {account}",
|
||
"",
|
||
"(Token obtained via gcloud auth print-access-token)",
|
||
], wait=False)
|
||
else:
|
||
tui_message(stdscr, [
|
||
"No active gcloud session found.",
|
||
"",
|
||
"Run the following in your terminal, then return here:",
|
||
" gcloud auth login",
|
||
"",
|
||
"Or enter credentials manually on the next screen.",
|
||
])
|
||
creds = tui_credentials(stdscr)
|
||
if creds is None:
|
||
return
|
||
account, token = creds
|
||
|
||
tui_message(stdscr, [
|
||
f"Authenticated as: {account}",
|
||
"",
|
||
"Fetching GCP organizations…",
|
||
], wait=False)
|
||
|
||
summary: dict = {}
|
||
orgs = fetch_gcp_orgs(token=token)
|
||
|
||
if not orgs:
|
||
tui_message(stdscr, [
|
||
f"No organizations found for {account}.",
|
||
"",
|
||
"You may lack resourcemanager.organizations.list permission,",
|
||
"or this account belongs to no GCP org.",
|
||
"",
|
||
"Continuing to project selection without an org filter.",
|
||
])
|
||
else:
|
||
sel = tui_list(stdscr, "Select GCP Organization", orgs,
|
||
subtitle=f"Authenticated as: {account}")
|
||
if sel is None:
|
||
return
|
||
summary["org_id"] = orgs[sel]["org_id"]
|
||
summary["org_name"] = orgs[sel]["org_name"]
|
||
|
||
tui_message(stdscr, ["Fetching projects…"], wait=False)
|
||
projects = fetch_gcp_projects(summary.get("org_id"), token=token)
|
||
|
||
if not projects:
|
||
tui_message(stdscr, [
|
||
"No projects found.",
|
||
"Ensure you have resourcemanager.projects.list permission.",
|
||
])
|
||
return
|
||
|
||
sel = tui_list(
|
||
stdscr,
|
||
"Select GCP Project",
|
||
projects,
|
||
subtitle=(
|
||
f"Org: {summary.get('org_name', summary.get('org_id', 'none'))}"
|
||
f" ({len(projects)} projects)"
|
||
),
|
||
)
|
||
if sel is None:
|
||
return
|
||
summary["project_id"] = projects[sel]["project_id"]
|
||
summary["project_name"] = projects[sel]["project_name"]
|
||
summary["project_number"] = projects[sel]["project_number"]
|
||
|
||
tui_message(stdscr, ["Fetching billing accounts…"], wait=False)
|
||
billing = fetch_gcp_billing_accounts(token=token)
|
||
|
||
if not billing:
|
||
tui_message(stdscr, [
|
||
"No open billing accounts found (or insufficient permissions).",
|
||
"",
|
||
"billing_account will be left blank in the output file.",
|
||
"Edit the file manually to fill it in.",
|
||
])
|
||
else:
|
||
sel = tui_list(stdscr, "Select Billing Account",
|
||
billing,
|
||
subtitle=f"Project: {summary['project_id']}")
|
||
if sel is None:
|
||
return
|
||
summary["billing_account_id"] = billing[sel]["billing_account_id"]
|
||
summary["billing_name"] = billing[sel]["billing_name"]
|
||
|
||
final_path = tui_confirm(stdscr, summary, output_path, mode, provider)
|
||
if final_path is None:
|
||
tui_message(stdscr, ["Cancelled. No file written."])
|
||
return
|
||
|
||
try:
|
||
write_gcp_config(final_path, summary, mode, provider)
|
||
except Exception as e:
|
||
tui_message(stdscr, [f"ERROR writing file: {e}"])
|
||
return
|
||
|
||
tui_message(stdscr, [
|
||
"Configuration written.",
|
||
"",
|
||
f" {final_path}",
|
||
"",
|
||
"Next steps:",
|
||
" 1. Copy org_id / billing_account / billing_project into:",
|
||
" deploy/gcp/terraform/cloud-setup.auto.tfvars",
|
||
" 2. Open the installer — Prod Cluster → Cloud tab will be pre-filled",
|
||
" (or click 'Load from gcp.cfg' to refresh on demand)",
|
||
])
|
||
|
||
|
||
# -- CLI entry point ---------------------------------------------------------
|
||
|
||
def _gcp_config_main():
|
||
parser = argparse.ArgumentParser(
|
||
prog="config.sh",
|
||
description=(
|
||
"ncurses TUI: select GCP credentials → write flat config file.\n"
|
||
"Output is used by the installer Prod Cluster screen and deploy pipelines."
|
||
),
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
epilog=(
|
||
"Examples:\n"
|
||
" ./config.sh --mode k8s --provider gcp\n"
|
||
" ./config.sh --mode prod --provider gcp --output /tmp/gcp.cfg\n"
|
||
),
|
||
)
|
||
parser.add_argument(
|
||
"--mode",
|
||
choices=SUPPORTED_MODES,
|
||
default="k8s",
|
||
help="Deployment mode: k8s or prod (synonymous). Default: k8s",
|
||
)
|
||
parser.add_argument(
|
||
"--provider",
|
||
choices=SUPPORTED_PROVIDERS,
|
||
default="gcp",
|
||
help="Cloud provider. Default: gcp",
|
||
)
|
||
parser.add_argument(
|
||
"--output",
|
||
default=str(DEFAULT_GCP_OUTPUT),
|
||
metavar="PATH",
|
||
help=f"Output file path. Default: {DEFAULT_GCP_OUTPUT}",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
mode = "k8s" if args.mode == "prod" else args.mode
|
||
|
||
if args.provider == "gcp":
|
||
try:
|
||
curses.wrapper(_run_gcp_tui, args.output, mode, args.provider)
|
||
except KeyboardInterrupt:
|
||
pass
|
||
except Exception as exc:
|
||
print(f"Fatal error: {exc}", file=sys.stderr)
|
||
sys.exit(1)
|
||
else:
|
||
print(f"Provider '{args.provider}' is not yet implemented.", file=sys.stderr)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
_gcp_config_main()
|