mirror of
https://github.com/dredx/prole.git
synced 2026-09-27 22:14:31 +00:00
455 lines
16 KiB
Python
455 lines
16 KiB
Python
"""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}")
|