"""Kerberos configuration, secret management and Ansible vault.""" import configparser import os import shutil import subprocess import tempfile import threading from pathlib import Path import tkinter as tk from tkinter import ttk, messagebox, filedialog from knoe import screen as ui from knoe.core.env import ( PROJECT_ROOT, _bool_str, _deployment_target_label, _ensure_ansible_vault_credentials, ) from knoe.config import ( SECRET_KEY_SPECS, _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, "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", ) self._render_title("Knoe Authority", 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 Knoe Authority", 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, 250, 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) left_col = content_width // 3 right_col_x = left_col + 16 right_col_w = content_width - right_col_x - 48 x_field = 160 field_w = left_col - x_field - 8 y += 60 console_y = y 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) self._kerberos_console = self._create_console_output( x=right_col_x, y=console_y, title="Kerberos Output", width=right_col_w, height=350 ) 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"), self.next_button.configure(state="normal"), ) ) 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 @staticmethod def _yaml_scalar(content: str, key: str) -> str: prefix = f"{key}:" for line in content.splitlines(): stripped = line.strip() if not stripped or stripped.startswith("#"): continue if not stripped.startswith(prefix): continue value = stripped[len(prefix) :].strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in ('"', "'"): value = value[1:-1] return value return "" @staticmethod def _vault_password_files() -> list[Path]: files: list[Path] = [] env_pw = os.environ.get("ANSIBLE_VAULT_PASSWORD_FILE", "").strip() if env_pw: files.append(Path(env_pw)) for base in (Path.cwd(), PROJECT_ROOT): files.append(base / ".vault_pass") seen: set[Path] = set() unique: list[Path] = [] for item in files: if item in seen: continue seen.add(item) unique.append(item) return unique def _resolve_samba_admin_password_from_vault(self) -> str: vault_path = ( PROJECT_ROOT / "infrastructure" / "inventory" / "group_vars" / "ad_dc" / "vault.yml" ) if not vault_path.exists(): return "" try: raw = vault_path.read_text(encoding="utf-8") except Exception: raw = "" if raw and "$ANSIBLE_VAULT;" not in raw: return self._yaml_scalar(raw, "vault_samba_dns_admin_pass") if shutil.which("ansible-vault") is None: return "" _ensure_ansible_vault_credentials(prompt_ui=False) for password_file in self._vault_password_files(): if not password_file.is_file(): continue try: res = subprocess.run( [ "ansible-vault", "view", str(vault_path), "--vault-password-file", str(password_file), ], capture_output=True, text=True, check=False, ) except Exception: continue if res.returncode != 0: continue value = self._yaml_scalar(res.stdout or "", "vault_samba_dns_admin_pass") if value: return value return "" 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 _is_openbao_ref(db_val): # Attempt to resolve from OpenBao for the UI try: db_plain = self._resolve_openbao_ref(db_val) if db_plain == db_val: # Resolution failed db_plain = "" except Exception: db_plain = "" elif 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 _is_openbao_ref(krb_val): # Attempt to resolve from OpenBao for the UI try: krb_plain = self._resolve_openbao_ref(krb_val) if krb_plain == krb_val: # Resolution failed krb_plain = "" except Exception: krb_plain = "" elif krb_val: krb_plain = krb_val else: krb_plain = "" if not krb_plain: krb_plain = self._resolve_samba_admin_password_from_vault() 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["KNOE_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}")