chrisfu/prole
1
0
Fork 0
mirror of https://github.com/dredx/prole.git synced 2026-09-24 18:24:32 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity
64e1cef16c
prole/knoe/ui/screens/security.py
chrisfu 3312c39b1f feat: GCP/GKE CNPG hardening, Artifact Registry traffic light, and knoe-system namespace fixes 
UI screens
- database.py: fix mode detection to use env_key priority (prod→k8s, service→k3s) so stale DEPLOYMENT_MODE never overrides the user's chosen environment
- database.py: Registry status reads ARTIFACT_REGISTRY_AVAILABLE persisted by cluster screen; uses SERVICE_NAMESPACE for Artifact Registry repo name
- cluster.py: add Artifact Registry traffic light (amber→green/red) to prod section; _check_artifact_registry_async persists ARTIFACT_REGISTRY_AVAILABLE into Global cfg
- cluster.py: re-trigger Artifact Registry check after GKE cluster selection so the light re-evaluates once region is available from KUBECONTEXT
- cluster_nodes.py: fix TclError on Python 3.14 — pady=(2,0) tuple → pady=2 scalar
- __init__.py: seed knoe-system namespace when saved value is "default", not only when empty
- services.py: replace hardcoded "Prole DB" log string with dynamic cnpg_cluster name

Core ops
- cloudnative_pg.py: replace one-shot Barman plugin retry with 6-attempt loop; first cert-manager/x509 failure triggers rollout restart + 30 s CA propagation wait; subsequent failures back off up to 60 s per attempt
- cloudnative_pg.py: TLS CA CN now uses cluster_name instead of hardcoded "Prole CNPG CA"
- registry.py, garage_store.py: refactored into per-mode modules (k3d/k3s/k8s registry and garage store, shared _garage_common)

Deploy / config
- deploy/gcp/gke/knoe-db.yaml: GKE-specific CNPG cluster manifest (rw/ro/r on separate nodes with premium-rwo storage)
- etc/init_common_services.sh, modes/k8s/knoe-db/.version: updated for current deploy
- kong-deployment.yaml: updated manifest

Tests
- test_cluster_nodes_render_smoke.py: add pack/grid, winfo_children, winfo_reqheight, update_idletasks, grid_slaves to dummy widgets; monkeypatch tk.Label so CNPG placement render completes without a real Tkinter root

Co-authored-by: Junie <junie@jetbrains.com>
2026-04-04 19:36:08 -07:00

706 lines
24 KiB
Python
Raw Blame History

"""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}")
Reference in New Issue View Git Blame Copy Permalink