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 <noreply@anthropic.com>
This commit is contained in:
chrisfu 2026-03-31 23:11:26 -07:00
parent 93f7259722
commit 2e517686a8
4 changed files with 525 additions and 0 deletions

1
.gitignore vendored
View File

@ -78,3 +78,4 @@ htmlcov/
/prole-auth/target/surefire-reports/org.prole.auth.web.VerifyControllerTest.txt /prole-auth/target/surefire-reports/org.prole.auth.web.VerifyControllerTest.txt
/prole-db.iml /prole-db.iml
supabase/helm/generated/values.generated.json supabase/helm/generated/values.generated.json
/.claude/

7
config.sh Executable file
View File

@ -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 "$@"

View File

@ -62,6 +62,7 @@
<excludeFolder url="file://$MODULE_DIR$/ssh-keys" /> <excludeFolder url="file://$MODULE_DIR$/ssh-keys" />
<excludeFolder url="file://$MODULE_DIR$/tmp" /> <excludeFolder url="file://$MODULE_DIR$/tmp" />
<excludeFolder url="file://$MODULE_DIR$/vault_backup" /> <excludeFolder url="file://$MODULE_DIR$/vault_backup" />
<excludeFolder url="file://$MODULE_DIR$/.claude" />
</content> </content>
</component> </component>
<component name="FacetManager"> <component name="FacetManager">

View File

@ -7,7 +7,9 @@ package per the refactor request. UI code should import from `knoe.config`.
from __future__ import annotations from __future__ import annotations
import argparse
import base64 import base64
import curses
import getpass import getpass
import json import json
import os import os
@ -1241,3 +1243,517 @@ def _parse_gcp_cfg(path) -> dict:
pass pass
return result 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) -> 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()