"""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): # 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, 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) 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 # Use tk.Text self.kerberos_status_text = tk.Text(self.bg_canvas, width=100, height=12, font=('Menlo', 10), bg='white', fg='black', insertbackground='black', highlightbackground='#CCCCCC', highlightthickness=1, relief='flat') status_window = self.bg_canvas.create_window(48, y, window=self.kerberos_status_text, anchor='nw', width=800, height=200) self._canvas_items.append(status_window) self._overlay_widgets.append(self.kerberos_status_text) 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 self.kerberos_status_text.delete('1.0', tk.END) self.kerberos_status_text.insert(tk.END, f"Testing {user}@{realm} inside the cluster...\n") 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): def _do_write(l=line): try: if self.kerberos_status_text.winfo_exists(): self.kerberos_status_text.insert(tk.END, l) self.kerberos_status_text.see(tk.END) except (tk.TclError, RuntimeError): pass self.safe_after(_do_write) 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") else: write_line(f"\nKerberos test failed with code {rc2}\n") except Exception as e: def _do_error(msg=str(e)): try: if self.kerberos_status_text.winfo_exists(): self.kerberos_status_text.insert(tk.END, f"Error: {msg}\n") except (tk.TclError, RuntimeError): pass self.safe_after(_do_error) 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 _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}")