From 2e517686a80c3802bbbb1d964c7eff33eeeb6761 Mon Sep 17 00:00:00 2001 From: chrisfu Date: Tue, 31 Mar 2026 23:11:26 -0700 Subject: [PATCH] Add GCP ncurses TUI to knoe/config.py and config.sh launcher Consolidates the GCP cluster configuration TUI (formerly etc/config.py) into knoe/config.py as a __main__ entrypoint, resolving the naming collision with the root-level installer utilities module. Adds config.sh as a thin shell launcher at the project root. Also excludes .claude/ worktree directories from git and IDE indexing. Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 1 + config.sh | 7 + knoe-db.iml | 1 + knoe/config.py | 516 +++++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 525 insertions(+) create mode 100755 config.sh diff --git a/.gitignore b/.gitignore index 95da67f..fb151ab 100644 --- a/.gitignore +++ b/.gitignore @@ -78,3 +78,4 @@ htmlcov/ /prole-auth/target/surefire-reports/org.prole.auth.web.VerifyControllerTest.txt /prole-db.iml supabase/helm/generated/values.generated.json +/.claude/ diff --git a/config.sh b/config.sh new file mode 100755 index 0000000..ad21325 --- /dev/null +++ b/config.sh @@ -0,0 +1,7 @@ +#!/usr/bin/env bash +# config.sh — GCP cluster configuration TUI launcher +# Wraps: python3 -m knoe.config --mode k8s --provider gcp +set -euo pipefail +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +cd "$SCRIPT_DIR" +exec python3 -m knoe.config "$@" diff --git a/knoe-db.iml b/knoe-db.iml index 9bd88ac..c72f05c 100644 --- a/knoe-db.iml +++ b/knoe-db.iml @@ -62,6 +62,7 @@ + diff --git a/knoe/config.py b/knoe/config.py index de13655..fbf0b20 100644 --- a/knoe/config.py +++ b/knoe/config.py @@ -7,7 +7,9 @@ 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 @@ -1241,3 +1243,517 @@ def _parse_gcp_cfg(path) -> dict: 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 = "" +# billing_account = "" +# billing_project = "" +# project_id = "" +# --------------------------------------------------------------------------- + +DEFAULT_GCP_OUTPUT = PROJECT_ROOT / "conf" / "prod" / "gcp.cfg" +SUPPORTED_MODES = ("k8s", "prod") +SUPPORTED_PROVIDERS = ("gcp",) + + +# -- gcloud helpers ---------------------------------------------------------- + +def _gcloud(*args) -> tuple[int, list | dict]: + cmd = ["gcloud", *args, "--format=json", "--quiet"] + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + 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) -> tuple[int, str]: + cmd = ["gcloud", *args, "--quiet"] + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + 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() -> str | None: + rc, data = _gcloud("auth", "list", "--filter=status=ACTIVE") + if rc == 0 and data: + return data[0].get("account", "") + return None + + +def fetch_gcp_orgs() -> list[dict]: + rc, data = _gcloud("organizations", "list") + 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) -> list[dict]: + args = ["projects", "list"] + if org_id: + args += [f"--filter=parent.id={org_id} AND parent.type=organization"] + rc, data = _gcloud(*args) + 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() -> list[dict]: + rc, data = _gcloud("beta", "billing", "accounts", "list", "--filter=open=true") + if rc != 0 or not data: + rc, data = _gcloud("billing", "accounts", "list", "--filter=open=true") + 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/prod/prole.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") + + +# -- 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 + + tui_message(stdscr, ["Checking gcloud authentication…"], wait=False) + account = active_gcp_account() + + if not account: + tui_message(stdscr, [ + "No active gcloud account found.", + "", + "This will run: gcloud auth login --no-browser", + "", + "A device-code URL will be printed to the terminal.", + "Open it in any browser (on any machine) to authenticate.", + "", + "Press any key to begin, or q to quit.", + ]) + k = stdscr.getch() + if k in (ord("q"), ord("Q"), 27): + return + + curses.endwin() + print("\nRunning: gcloud auth login --no-browser\n") + rc, _ = _gcloud_plain("auth", "login", "--no-browser") + stdscr = curses.initscr() + curses.start_color() + curses.use_default_colors() + stdscr.keypad(True) + + if rc != 0: + tui_message(stdscr, ["Login failed or was cancelled. Exiting."]) + return + + account = active_gcp_account() + if not account: + tui_message(stdscr, ["Auth succeeded but no active account detected. Exiting."]) + return + + tui_message(stdscr, [ + f"Authenticated as: {account}", + "", + "Fetching GCP organizations…", + ], wait=False) + + summary: dict = {} + orgs = fetch_gcp_orgs() + + 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")) + + 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() + + 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()