"""Kerberos configuration, secret management and Ansible vault.""" import configparser import os import shutil import subprocess import threading from pathlib import Path import tkinter as tk from tkinter import ttk, messagebox, filedialog from installer import screen as ui from installer.core.env import ( PROJECT_ROOT, _bool_str, _deployment_target_label, _ensure_ansible_vault_credentials, ) from installer.config import _decrypt_prole_secret, _is_openbao_ref, _is_prole_secret class SecurityScreenMixin: """Kerberos configuration, secret management and Ansible vault.""" def _render_kerberos_config_page(self): self._kerberos_success = False initial_status = "Ready" if self.kerberos_enabled.get() else "Disabled" self._kerberos_status_var = tk.StringVar(value=initial_status) # Letterhead at top right content_width = self.bg_canvas.winfo_width() or 975 right_margin = content_width - 48 ui.canvas_text( self, right_margin, 40, "Prole", fill="#6e6e73", font=("SF Pro Text", 32, "bold"), anchor="ne", ) ui.canvas_text( self, right_margin, 85, "Infrastructure Automated.", fill="#6e6e73", font=("SF Pro Text", 18), anchor="ne", ) self._render_title("Kerberos Authentication", y=150) self._render_paragraph( "Configure Kerberos authentication for Prole and the PostgreSQL database. Tests run inside the Kubernetes namespace to validate against your realm.", y=200, ) y = 280 # Use tk.Checkbutton on canvas enable_cb = tk.Checkbutton( self.bg_canvas, text="Enable Kerberos Authentication", variable=self.kerberos_enabled, command=self._on_kerberos_toggle, bg="white", fg="black", activebackground="white", selectcolor="white", font=("SF Pro Text", 11), ) cb_window = self.bg_canvas.create_window(48, y, window=enable_cb, anchor="nw") self._canvas_items.append(cb_window) self._overlay_widgets.append(enable_cb) status_item = ui.canvas_text( self, 340, y + 6, f"Status: {initial_status}", fill="black", font=("SF Pro Text", 11), ) def _update_status_text(*_): try: self.bg_canvas.itemconfig( status_item, text=f"Status: {self._kerberos_status_var.get()}" ) except Exception: pass self._kerberos_status_var.trace_add("write", _update_status_text) x_field = 200 field_w = 400 y += 60 ui.canvas_text( self, 48, y, "Kerberos Realm:", fill="black", font=("SF Pro Text", 12) ) realm_entry = tk.Entry( self.bg_canvas, textvariable=self.kerberos_realm, bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) realm_window = self.bg_canvas.create_window( x_field, y - 4, window=realm_entry, anchor="nw", width=field_w, height=32 ) self._canvas_items.append(realm_window) self._overlay_widgets.append(realm_entry) y += 48 ui.canvas_text( self, 48, y, "KDC Host/IP:", fill="black", font=("SF Pro Text", 12) ) kdc_entry = tk.Entry( self.bg_canvas, textvariable=self.kerberos_kdc, bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) kdc_window = self.bg_canvas.create_window( x_field, y - 4, window=kdc_entry, anchor="nw", width=field_w, height=32 ) self._canvas_items.append(kdc_window) self._overlay_widgets.append(kdc_entry) y += 48 ui.canvas_text(self, 48, y, "Username:", fill="black", font=("SF Pro Text", 12)) user_entry = tk.Entry( self.bg_canvas, textvariable=self.kerberos_user, bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) user_window = self.bg_canvas.create_window( x_field, y - 4, window=user_entry, anchor="nw", width=field_w, height=32 ) self._canvas_items.append(user_window) self._overlay_widgets.append(user_entry) y += 48 ui.canvas_text(self, 48, y, "Password:", fill="black", font=("SF Pro Text", 12)) pass_entry = tk.Entry( self.bg_canvas, textvariable=self.kerberos_password, show="*", bg="white", fg="black", insertbackground="black", highlightbackground="#CCCCCC", highlightthickness=1, relief="flat", font=("SF Pro Text", 11), ) pass_window = self.bg_canvas.create_window( x_field, y - 4, window=pass_entry, anchor="nw", width=field_w, height=32 ) self._canvas_items.append(pass_window) self._overlay_widgets.append(pass_entry) y += 60 # Use tk.Button on canvas test_btn = tk.Button( self.bg_canvas, text="Test Connection", command=self.test_kerberos_connection, bg="#F5F5DC", fg="black", activebackground="#E5E5D5", highlightbackground="#F5F5DC", highlightthickness=0, relief="flat", font=("SF Pro Text", 11), padx=16, pady=8, ) btn_window = self.bg_canvas.create_window(48, y, window=test_btn, anchor="nw") self._canvas_items.append(btn_window) self._overlay_widgets.append(test_btn) y += 60 self._kerberos_console = self._create_console_output( y=y, title="Kerberos Output", width=900, height=260 ) def test_kerberos_connection(self): self._action_flags["kerberos_config.test_connection"] = True realm = self.kerberos_realm.get().strip() user = self.kerberos_user.get().strip() password = self.kerberos_password.get().strip() kdc = self.kerberos_kdc.get().strip() if not realm or not user or not password or not kdc: messagebox.showerror( "Error", "Please fill in realm, KDC host, username, and password." ) return try: self.prole_cfg_data["Kerberos Authentication"]["ENABLED"] = str( self.kerberos_enabled.get() ) self.prole_cfg_data["Kerberos Authentication"]["REALM"] = realm self.prole_cfg_data["Kerberos Authentication"]["KDC"] = kdc self.prole_cfg_data["Kerberos Authentication"]["SERVER"] = kdc self.prole_cfg_data["Kerberos Authentication"]["USER"] = user self.prole_cfg_data["Kerberos Authentication"]["PASSWORD"] = password self.prole_cfg_data["Optional Features"]["KERBEROS_ENABLED"] = str( self.kerberos_enabled.get() ) self.prole_cfg_data["Kerberos Authentication"]["AD_PORT_FORWARD"] = ( os.environ.get("KRB5_AD_PORT_FORWARD", "1") ) self.prole_cfg_data["Kerberos Authentication"]["AD_TCP_PORTS"] = ( os.environ.get("KRB5_AD_TCP_PORTS", "88 389 445 464 636") ) self.prole_cfg_data["Kerberos Authentication"]["AD_UDP_PORTS"] = ( os.environ.get("KRB5_AD_UDP_PORTS", "88 464") ) self.prole_cfg_data["Kerberos Authentication"]["AD_PROXY_HOST_NETWORK"] = ( os.environ.get("KRB5_AD_PROXY_HOST_NETWORK", "1") ) self.prole_cfg_data["Kerberos Authentication"]["AD_PROXY_IMAGE"] = ( os.environ.get("KRB5_AD_PROXY_IMAGE", "alpine/socat") ) self.prole_cfg_data["Kerberos Authentication"]["AD_PROXY_SERVICE"] = ( os.environ.get("KRB5_AD_SERVICE_NAME", "prole-kerberos-ad-dc") ) self._save_prole_cfg() except Exception: pass try: self._kerberos_console.clear() self._kerberos_console.write( f"Testing {user}@{realm} inside the cluster...\n" ) except Exception: pass def worker(): try: env = os.environ.copy() env["PROLE_HOME"] = str(PROJECT_ROOT) env["PROLE_SERVICE"] = str(PROJECT_ROOT) env["NAMESPACE"] = (self.db_namespace.get() or "").strip() env["SERVICE_NAMESPACE"] = self._get_service_namespace() env["KRB5_REALM"] = realm env["REALM"] = realm env["DOMAIN"] = realm.lower() env["KRB5_KDC"] = kdc env["KRB5_ADMIN"] = kdc env["KRB5_USER"] = user env["KRB5_PASSWORD"] = password env["KERBEROS_ENABLED"] = _bool_str(self.kerberos_enabled.get()) env["ENABLED"] = env["KERBEROS_ENABLED"] mode = self._deployment_mode() if mode == "k3s": env.setdefault("KRB5_AD_PORT_FORWARD", "0") if mode: env["PROLE_MODE"] = mode env["DEPLOYMENT_MODE"] = mode env["DEPLOYMENT_TARGET"] = _deployment_target_label( self.cluster_env.get() ) mode_args = ["--mode", mode] if mode else [] def write_line(line): try: self._kerberos_console.write(line) except Exception: pass write_line("\n==> init_kerberos.sh test\n") rc2 = self.controller.run_script( "init_kerberos.sh", args=mode_args + ["test"], env=env, on_line=write_line, ) if rc2 == 0: write_line("\nKerberos test completed successfully.\n") self.safe_after( lambda: self._kerberos_status_var.set("Test Passed") ) else: write_line(f"\nKerberos test failed with code {rc2}\n") self.safe_after( lambda: self._kerberos_status_var.set( f"Test Failed (Code {rc2})" ) ) except Exception as e: try: self._kerberos_console.write(f"Error: {e}\n") except Exception: pass threading.Thread(target=worker, daemon=True).start() def _secret_namespace(self) -> str: ns = "" try: ns = (self.db_namespace.get() or "").strip() except Exception: ns = "" if not ns: ns = os.environ.get("NAMESPACE", "") or "default" return ns def _load_secret_cache_from_cfg(self): try: if self._cfg_path_override is not None: cfg_path = self._cfg_path_override if cfg_path.is_dir(): cfg_path = cfg_path / "prole.cfg" else: conf_dir = self._resolve_prole_conf_dir() cfg_path = conf_dir / "prole.cfg" except Exception: return if not cfg_path.exists(): return cfg = configparser.ConfigParser(interpolation=None) cfg.optionxform = str cfg.read(cfg_path) for (section, key), _spec in SECRET_KEY_SPECS.items(): if cfg.has_option(section, key): val = cfg.get(section, key, fallback="").strip() if val: self._cfg_secret_cache[(section, key)] = val if _is_openbao_ref(val): self._secrets_finalized = True def _resolve_cached(pairs: list[tuple[str, str]]) -> str: for pair in pairs: val = self._cfg_secret_cache.get(pair, "") if val: return val return "" db_val = _resolve_cached( [ ("Inputs", "init_password.db_password"), ("Global", "DB_PASSWORD"), ] ) if db_val and _is_prole_secret(db_val): db_plain = _decrypt_prole_secret(db_val) elif db_val and not _is_openbao_ref(db_val): db_plain = db_val else: db_plain = "" if db_plain: try: self.db_password.set(db_plain) self.db_password_confirm.set(db_plain) except Exception: pass krb_val = _resolve_cached( [ ("Inputs", "kerberos_config.password"), ("Kerberos Authentication", "PASSWORD"), ] ) if krb_val and _is_prole_secret(krb_val): krb_plain = _decrypt_prole_secret(krb_val) elif krb_val and not _is_openbao_ref(krb_val): krb_plain = krb_val else: krb_plain = "" if krb_plain: try: self.kerberos_password.set(krb_plain) except Exception: pass def _kerberos_script_env(self): env = os.environ.copy() env["PROLE_HOME"] = str(PROJECT_ROOT) env["PROLE_SERVICE"] = str(PROJECT_ROOT) env["PROLE_DB_USER"] = self.db_username.get() env["DB_PASSWORD"] = self.db_password.get() env["NAMESPACE"] = (self.db_namespace.get() or "").strip() or "default" env["SERVICE_NAMESPACE"] = self._get_service_namespace() realm = self.kerberos_realm.get().strip() if realm: env["KRB5_REALM"] = realm env["REALM"] = realm env["DOMAIN"] = realm.lower() kdc = self.kerberos_kdc.get().strip() if kdc: env["KRB5_KDC"] = kdc env["KRB5_ADMIN"] = kdc user = self.kerberos_user.get().strip() if user: env["KRB5_USER"] = user password = self.kerberos_password.get().strip() if password: env["KRB5_PASSWORD"] = password enabled = _bool_str(self.kerberos_enabled.get()) env["KERBEROS_ENABLED"] = enabled env["ENABLED"] = enabled mode = self._deployment_mode() mode_args = ["--mode", mode] if mode else [] if mode: env["PROLE_MODE"] = mode env["DEPLOYMENT_MODE"] = mode env["DEPLOYMENT_TARGET"] = _deployment_target_label( self.cluster_env.get() ) return env, mode_args def _run_kerberos_init(self): if getattr(self, "_kerberos_init_running", False): return if not self.kerberos_enabled.get(): self._kerberos_status_var.set("Disabled") return self._kerberos_init_running = True self._kerberos_status_var.set("Initializing...") try: self._kerberos_console.clear() self._kerberos_console.write("Running init_kerberos.sh initialize...\n") except Exception: pass self.update_footer() def worker(): env, mode_args = self._kerberos_script_env() script_path = PROJECT_ROOT / "etc" / "init_kerberos.sh" if not script_path.exists(): try: self._kerberos_console.write( f"Error: {script_path} not found.\n" ) self._kerberos_status_var.set("Failed (Script not found)") except Exception: pass self._kerberos_init_running = False self.safe_after(self.update_footer) return rc = self.controller.run_script( "init_kerberos.sh", args=mode_args + ["initialize"], env=env, on_line=lambda l: self._kerberos_console.write(l), ) if rc == 0: self._kerberos_success = True self.safe_after( lambda: self._kerberos_status_var.set("Initialized") ) try: self._kerberos_console.write( "\nKerberos initialization completed successfully.\n" ) except Exception: pass else: self._kerberos_success = False self.safe_after( lambda: self._kerberos_status_var.set(f"Failed (Code {rc})") ) try: self._kerberos_console.write( f"\nKerberos initialization failed with exit code {rc}.\n" ) except Exception: pass self._kerberos_init_running = False self.safe_after(self.update_footer) threading.Thread(target=worker, daemon=True).start() def _on_kerberos_toggle(self): enabled = self.kerberos_enabled.get() try: self.prole_cfg_data["Optional Features"]["KERBEROS_ENABLED"] = str(enabled) self._save_prole_cfg() except Exception: pass if not enabled: self._kerberos_success = False self._kerberos_status_var.set("Disabled") try: self._kerberos_console.write("Kerberos disabled. Skipping init.\n") except Exception: pass self.update_footer() return self._kerberos_status_var.set("Enabled") try: self._kerberos_console.write("Kerberos enabled. Starting initialization...\n") except Exception: pass self.update_footer() self.safe_after(self._run_kerberos_init) def _save_ansible_prole_vault(self, root_password: str): if not root_password: return vault_path = ( PROJECT_ROOT / "infrastructure" / "inventory" / "group_vars" / "all" / "prole_vault.yml" ) _ensure_ansible_vault_credentials(prompt_ui=False) vault_pass_file = os.environ.get("ANSIBLE_VAULT_PASSWORD_FILE") if not vault_pass_file or not Path(vault_pass_file).exists(): for base in (Path.cwd(), PROJECT_ROOT): candidate = base / ".vault_pass" if candidate.is_file(): vault_pass_file = str(candidate) break if not vault_pass_file or not Path(vault_pass_file).exists(): return if shutil.which("ansible-vault") is None: return try: # We must decrypt if it is a prole secret plain_pass = root_password if _is_prole_secret(root_password): try: plain_pass = _decrypt_prole_secret(root_password) except Exception: pass content = f'prole_root_password: "{plain_pass}"\n' with tempfile.NamedTemporaryFile(mode="w", delete=False) as f: f.write(content) tmp_name = f.name try: os.chmod(tmp_name, 0o600) subprocess.run( [ "ansible-vault", "encrypt", tmp_name, "--vault-password-file", vault_pass_file, ], check=True, capture_output=True, ) shutil.move(tmp_name, str(vault_path)) os.chmod(str(vault_path), 0o600) finally: if os.path.exists(tmp_name): try: os.unlink(tmp_name) except: pass except Exception as e: print(f"[ERROR] Failed to save ansible vault: {e}")