"""Kerberos configuration and secret management.""" import configparser import os import threading 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, ) from knoe.config import ( SECRET_KEY_SPECS, _decrypt_knoe_secret, _is_openbao_ref, _is_knoe_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 Knoe 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.knoe_cfg_data["Kerberos Authentication"]["ENABLED"] = str( self.kerberos_enabled.get() ) self.knoe_cfg_data["Kerberos Authentication"]["REALM"] = realm self.knoe_cfg_data["Kerberos Authentication"]["KDC"] = kdc self.knoe_cfg_data["Kerberos Authentication"]["SERVER"] = kdc self.knoe_cfg_data["Kerberos Authentication"]["USER"] = user self.knoe_cfg_data["Kerberos Authentication"]["PASSWORD"] = password self.knoe_cfg_data["Optional Features"]["KERBEROS_ENABLED"] = str( self.kerberos_enabled.get() ) self.knoe_cfg_data["Kerberos Authentication"]["AD_PORT_FORWARD"] = ( os.environ.get("KRB5_AD_PORT_FORWARD", "1") ) self.knoe_cfg_data["Kerberos Authentication"]["AD_TCP_PORTS"] = ( os.environ.get("KRB5_AD_TCP_PORTS", "88 389 445 464 636") ) self.knoe_cfg_data["Kerberos Authentication"]["AD_UDP_PORTS"] = ( os.environ.get("KRB5_AD_UDP_PORTS", "88 464") ) self.knoe_cfg_data["Kerberos Authentication"]["AD_PROXY_HOST_NETWORK"] = ( os.environ.get("KRB5_AD_PROXY_HOST_NETWORK", "1") ) self.knoe_cfg_data["Kerberos Authentication"]["AD_PROXY_IMAGE"] = ( os.environ.get("KRB5_AD_PROXY_IMAGE", "alpine/socat") ) self.knoe_cfg_data["Kerberos Authentication"]["AD_PROXY_SERVICE"] = ( os.environ.get("KRB5_AD_SERVICE_NAME", "knoe-kerberos-ad-dc") ) self._save_knoe_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["KNOE_HOME"] = str(PROJECT_ROOT) env["KNOE_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["KNOE_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 "" def _resolve_samba_admin_password_from_vault(self) -> str: try: from knoe.core.onepassword import get_secret, op_available if op_available(): return get_secret("samba-dns-admin", "password") except Exception: pass 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 / "knoe.cfg" else: conf_dir = self._resolve_knoe_conf_dir() cfg_path = conf_dir / "knoe.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_knoe_secret(db_val): db_plain = _decrypt_knoe_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_knoe_secret(krb_val): krb_plain = _decrypt_knoe_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["KNOE_HOME"] = str(PROJECT_ROOT) env["KNOE_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["KNOE_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.knoe_cfg_data["Optional Features"]["KERBEROS_ENABLED"] = str(enabled) self._save_knoe_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_knoe_vault(self, root_password: str): """Store the DB master password in the 1Password knoey vault.""" if not root_password: return try: plain_pass = root_password if _is_knoe_secret(root_password): try: plain_pass = _decrypt_knoe_secret(root_password) except Exception: pass from knoe.core.onepassword import set_secret, op_available if op_available(): set_secret("administrator", "password", plain_pass) except Exception as e: print(f"[ERROR] Failed to save administrator password to 1Password: {e}")