"""Knoe system user account provisioning screen. Appears after the Kerberos Authentication screen (when Kerberos is enabled). Runs etc/init_knoe_users.sh to: - Create admin@PROLE.LOCAL and guest@PROLE.LOCAL KDC principals - Export the postgres service keytab into a k8s Secret - Patch CNPG cluster with declarative managed roles (admin, guest, developer) - Create the demo schema and grants - Patch the ArgoCD RBAC policy to grant admin the ArgoCD admin role - Promote the admin user in Gitea / GitLab via API (optional, token-gated) """ from __future__ import annotations import os import subprocess import threading import tkinter as tk from knoe import screen as ui from knoe.core.env import PROJECT_ROOT class KnoeUsersScreenMixin: """Knoe system account provisioning screen.""" def _render_knoe_users_page(self): self._knoe_users_success = False self._knoe_users_deploying = False self._knoe_users_status_var = tk.StringVar(value="Ready") content_width = self.bg_canvas.winfo_width() or 975 right_margin = content_width - 48 # ── Letterhead ──────────────────────────────────────────────────────── ui.canvas_text( self, right_margin, 40, "knoe.dev", fill="#6e6e73", font=("SF Pro Text", 32, "bold"), anchor="ne", ) ui.canvas_text( self, right_margin, 85, "infrastructure.auto()", fill="#6e6e73", font=("SF Pro Text", 18), anchor="ne", ) # ── Title & description ─────────────────────────────────────────────── self._render_title("Knoe System Accounts", y=150) self._render_paragraph( "Provision Kerberos principals and database roles for the knoe-system " "cluster. Creates admin and guest accounts in the KDC, exports the " "postgres service keytab, patches CNPG managed roles (admin, guest, " "developer), initialises the demo schema, and wires admin into ArgoCD " "RBAC. Gitea / GitLab API promotion is optional and can be re-run " "after the admin user completes their first login.", y=205, ) # ── Status line ─────────────────────────────────────────────────────── status_item = ui.canvas_text( self, 48, 280, "Status: Ready", fill="black", font=("SF Pro Text", 12), ) def _update_status(*_): try: self.bg_canvas.itemconfig( status_item, text=f"Status: {self._knoe_users_status_var.get()}", ) except Exception: pass self._knoe_users_status_var.trace_add("write", _update_status) # ── Optional token fields ───────────────────────────────────────────── x_label = 48 x_field = 260 y = 320 ui.canvas_text( self, x_label, y, "Gitea Admin Token (optional):", fill="black", font=("SF Pro Text", 11), ) if not hasattr(self, "gitea_admin_token"): self.gitea_admin_token = tk.StringVar() gitea_entry = tk.Entry( self.bg_canvas, textvariable=self.gitea_admin_token, show="*", bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) gitea_window = self.bg_canvas.create_window( x_field, y - 4, window=gitea_entry, anchor="nw", width=380, height=28 ) self._canvas_items.append(gitea_window) self._overlay_widgets.append(gitea_entry) y += 40 ui.canvas_text( self, x_label, y, "GitLab Admin Token (optional):", fill="black", font=("SF Pro Text", 11), ) if not hasattr(self, "gitlab_admin_token"): self.gitlab_admin_token = tk.StringVar() gitlab_entry = tk.Entry( self.bg_canvas, textvariable=self.gitlab_admin_token, show="*", bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) gitlab_window = self.bg_canvas.create_window( x_field, y - 4, window=gitlab_entry, anchor="nw", width=380, height=28 ) self._canvas_items.append(gitlab_window) self._overlay_widgets.append(gitlab_entry) # ── Console output ──────────────────────────────────────────────────── self._knoe_users_console = self._create_console_output( y=415, title="Provisioning Output", width=content_width - 96, height=330 ) # ── Provision button ────────────────────────────────────────────────── self._knoe_users_btn = tk.Button( self.bg_canvas, text="Provision Accounts", command=self.run_knoe_users_provision, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", activeforeground="black", highlightbackground="#F5F5DC", highlightcolor="#F5F5DC", highlightthickness=0, relief="flat", bd=0, cursor="hand2", disabledforeground="#8B8B7A", font=("SF Pro Text", 11), padx=20, pady=10, ) btn_window = self.bg_canvas.create_window( 48, 760, window=self._knoe_users_btn, anchor="nw" ) self._canvas_items.append(btn_window) self._overlay_widgets.append(self._knoe_users_btn) # ── Script runner ───────────────────────────────────────────────────────── def run_knoe_users_provision(self): if getattr(self, "_knoe_users_deploying", False): return self._knoe_users_deploying = True self._knoe_users_success = False self._knoe_users_btn.configure(state="disabled") self._knoe_users_status_var.set("Provisioning…") self._knoe_users_console.clear() self.update_footer() def worker(): script_path = PROJECT_ROOT / "etc" / "init_knoe_users.sh" if not script_path.exists(): self._knoe_users_console.write( f"Error: {script_path} not found.\n" ) self.safe_after( lambda: self._knoe_users_status_var.set("Failed (script not found)") ) self.safe_after( lambda: self._knoe_users_btn.configure(state="normal") ) self._knoe_users_deploying = False return env = os.environ.copy() env["PROLE_HOME"] = str(PROJECT_ROOT) env["PROLE_SERVICE"] = str(PROJECT_ROOT) # Propagate namespace from installer state when available service_ns = ( getattr(self, "service_namespace", None) and self.service_namespace.get().strip() ) or os.environ.get("SERVICE_NAMESPACE", "knoe-system") env["SERVICE_NAMESPACE"] = service_ns env["KNOE_USERS_NAMESPACE"] = service_ns env["KNOE_KDC_NAMESPACE"] = service_ns db_ns = os.environ.get("DATABASE_NAMESPACE", "knoe-db") env["DATABASE_NAMESPACE"] = db_ns env["KNOE_DB_NAMESPACE"] = db_ns # Forward optional service admin tokens when provided gitea_token = ( getattr(self, "gitea_admin_token", None) and self.gitea_admin_token.get().strip() ) or "" if gitea_token: env["GITEA_ADMIN_TOKEN"] = gitea_token gitlab_token = ( getattr(self, "gitlab_admin_token", None) and self.gitlab_admin_token.get().strip() ) or "" if gitlab_token: env["GITLAB_ADMIN_TOKEN"] = gitlab_token # Pass prole.cfg path so the script sources the right config cfg_path = None try: if self._cfg_path_override and self._cfg_path_override.exists(): cfg_path = self._cfg_path_override else: conf_dir = self._resolve_prole_conf_dir() candidate = conf_dir / "prole.cfg" if candidate.exists(): cfg_path = candidate except Exception: cfg_path = None if cfg_path: env["PROLE_CFG"] = str(cfg_path) cmd = ["bash", str(script_path), "initialize"] self._knoe_users_console.write( f"Running: {' '.join(cmd)}\n" f"Namespace: {service_ns}\n\n" ) try: proc = subprocess.Popen( cmd, cwd=str(PROJECT_ROOT), env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1, ) except Exception as exc: self._knoe_users_console.write(f"Failed to start: {exc}\n") self.safe_after( lambda: self._knoe_users_status_var.set("Failed (launch error)") ) self.safe_after( lambda: self._knoe_users_btn.configure(state="normal") ) self._knoe_users_deploying = False return if proc.stdout: for line in iter(proc.stdout.readline, ""): if line: self._knoe_users_console.write(line) proc.stdout.close() rc = proc.wait() if rc == 0: self._knoe_users_success = True self.safe_after( lambda: self._knoe_users_status_var.set("Provisioned Successfully") ) self._knoe_users_console.write( "\nAccount provisioning completed successfully.\n" ) else: self._knoe_users_success = False self.safe_after( lambda: self._knoe_users_status_var.set(f"Failed (exit {rc})") ) self._knoe_users_console.write( f"\nProvisioning failed with exit code {rc}.\n" "Review the output above, resolve the issue, and click " "'Provision Accounts' again.\n" ) self._knoe_users_deploying = False self.safe_after(lambda: self._knoe_users_btn.configure(state="normal")) self.safe_after(self.update_footer) threading.Thread(target=worker, daemon=True).start()