"""Knoe User Authority screen. Builds the authority pod for the active deployment mode: - k3s / service → dev/prole/authority-prole-auth - prod / gcp → dev/prole/authority-gcp-auth The authority-prole-auth pod is wired for the primary KDC of the pod and the database store as intended. The database master password is used for admin@PROLE.LOCAL and the Kerberos admin credential is used for PROLE.ORG. """ 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 User Authority 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 User Authority", y=150) # Choose pod target label based on active mode _mode = (os.environ.get("PROLE_MODE") or "").strip().lower() _is_k3s = _mode in ("k3s", "service") or self._is_k3s_mode_active() _pod_target = "authority-prole-auth" if _is_k3s else "authority-gcp-auth" _pod_path = f"dev/prole/{_pod_target}" self._render_paragraph( f"Builds {_pod_path} for the current deployment mode " f"({'k3s' if _is_k3s else 'prod/gcp'}). " "The authority-prole-auth pod is wired for the primary KDC of the pod " "and the database store as intended. Provision Kerberos principals, " "export the postgres service keytab, patch CNPG managed roles (admin, " "guest, developer), initialise the demo schema, and wire admin into " "ArgoCD RBAC. The database master password is used for admin@PROLE.LOCAL; " "the Kerberos admin credential from the Authentication screen is used " "to connect PROLE.ORG.", y=205, ) # ── Status line ─────────────────────────────────────────────────────── status_item = ui.canvas_text( self, 48, 370, "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) # ── Console output ──────────────────────────────────────────────────── self._knoe_users_console = self._create_console_output( y=400, title="Provisioning Output", width=content_width - 96, height=355 ) # ── Provision button ────────────────────────────────────────────────── self._knoe_users_btn = tk.Button( self.bg_canvas, text="Build Authority Pod", 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): # noqa: C901 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(): # Select pod target and build script based on active mode _mode = (os.environ.get("PROLE_MODE") or "").strip().lower() _is_k3s = _mode in ("k3s", "service") or self._is_k3s_mode_active() _pod_target = "authority-prole-auth" if _is_k3s else "authority-gcp-auth" _pod_path = f"dev/prole/{_pod_target}" 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 # Authority pod target wired for primary KDC and database store env["AUTHORITY_POD_TARGET"] = _pod_target env["AUTHORITY_POD_PATH"] = _pod_path env["PROLE_AUTH_MODE"] = "k3s" if _is_k3s else "gcp" db_ns = os.environ.get("DATABASE_NAMESPACE", "knoe-db") env["DATABASE_NAMESPACE"] = db_ns env["KNOE_DB_NAMESPACE"] = db_ns # Use database master password for admin@PROLE.LOCAL db_pw = ( getattr(self, "db_password", None) and self.db_password.get().strip() ) or "" if db_pw: env["DB_MASTER_PASSWORD"] = db_pw env["PROLE_LOCAL_ADMIN_PASSWORD"] = db_pw # Use Kerberos admin credential from Authentication screen for PROLE.ORG krb_pw = ( getattr(self, "kerberos_password", None) and self.kerberos_password.get().strip() ) or "" if krb_pw: env["PROLE_ORG_ADMIN_PASSWORD"] = krb_pw env["KRB5_PASSWORD"] = krb_pw # 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"Building: {_pod_path}\n" 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( f"\n{_pod_path} built and provisioned 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"\nAuthority pod build failed with exit code {rc}.\n" "Review the output above, resolve the issue, and click " "'Build Authority Pod' 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()