# steward/settings/routes.py from __future__ import annotations import logging from pathlib import Path from quart import Blueprint, current_app, render_template, request, redirect, url_for, session from steward.auth.middleware import require_role from steward.core.audit import log_audit from steward.models.users import UserRole from steward.core.settings import ( get_all_settings, set_setting, to_smtp_cfg, to_webhook_cfg, to_ansible_cfg, to_plugins_cfg, to_oidc_cfg, to_ldap_cfg, to_thresholds_cfg, ) settings_bp = Blueprint("settings", __name__, url_prefix="/settings") logger = logging.getLogger(__name__) def _discover_plugins() -> list[dict]: """Scan every plugin root for plugin.yaml files. Roots are scanned in order (bundled first), and a plugin name found in an earlier root shadows the same name in a later one — matching load_plugins. """ import yaml plugin_dirs = [ Path(d).resolve() for d in current_app.config.get("PLUGIN_DIRS", ["plugins"]) ] plugins: list[dict] = [] seen: set[str] = set() for plugin_dir in plugin_dirs: if not plugin_dir.exists(): continue for entry in sorted(plugin_dir.iterdir()): yaml_path = entry / "plugin.yaml" if entry.is_dir() and yaml_path.exists() and entry.name not in seen: try: with yaml_path.open() as f: meta = yaml.safe_load(f) or {} meta["_dir"] = entry.name plugins.append(meta) seen.add(entry.name) except Exception: logger.warning("Could not read %s", yaml_path) return plugins def _merge_plugin_config(discovered: list[dict], plugins_cfg: dict) -> None: """Merge stored settings into discovered plugin dicts in-place.""" for plugin in discovered: name = plugin["_dir"] stored = plugins_cfg.get(name, {}) plugin["_enabled"] = stored.get("enabled", False) defaults = plugin.get("config", {}) plugin["_config"] = {**defaults, **{k: v for k, v in stored.items() if k != "enabled"}} async def _reload_app_config() -> None: """Reload app.config from DB so settings take effect without restart.""" async with current_app.db_sessionmaker() as db: fresh = await get_all_settings(db) current_app.config.update( SESSION_LIFETIME_HOURS=fresh.get("session.lifetime_hours", 8), DATA_RETENTION_DAYS=fresh.get("data.retention_days", 90), MONITORS_POLL_INTERVAL=fresh.get("monitors.poll_interval_seconds", 60), PUBLIC_BASE_URL=fresh.get("general.public_base_url", ""), SMTP=to_smtp_cfg(fresh), WEBHOOK=to_webhook_cfg(fresh), ANSIBLE=to_ansible_cfg(fresh), PLUGINS=to_plugins_cfg(fresh), OIDC=to_oidc_cfg(fresh), LDAP=to_ldap_cfg(fresh), THRESHOLDS=to_thresholds_cfg(fresh), ) # ── Redirect root to general tab ────────────────────────────────────────────── @settings_bp.get("/") @require_role(UserRole.admin) async def index(): return redirect(url_for("settings.general")) # ── General ─────────────────────────────────────────────────────────────────── @settings_bp.get("/general/") @require_role(UserRole.admin) async def general(): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) return await render_template("settings/general.html", settings=settings) @settings_bp.post("/general/") @require_role(UserRole.admin) async def save_general(): form = await request.form async with current_app.db_sessionmaker() as db: async with db.begin(): await set_setting(db, "general.public_base_url", (form.get("general.public_base_url", "") or "").strip()) await set_setting(db, "session.lifetime_hours", int(form.get("session.lifetime_hours", 8))) await set_setting(db, "data.retention_days", int(form.get("data.retention_days", 90))) await set_setting(db, "monitors.poll_interval_seconds", int(form.get("monitors.poll_interval_seconds", 60))) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "general"}) return redirect(url_for("settings.general")) # ── Monitoring thresholds ───────────────────────────────────────────────────── # (form field, setting key, is_float) for every tunable cutoff. _THRESHOLD_FIELDS = [ ("cpu_warn", "thresholds.cpu_warn", False), ("cpu_crit", "thresholds.cpu_crit", False), ("mem_warn", "thresholds.mem_warn", False), ("mem_crit", "thresholds.mem_crit", False), ("disk_warn", "thresholds.disk_warn", False), ("disk_crit", "thresholds.disk_crit", False), ("load_warn", "thresholds.load_warn", False), ("load_crit", "thresholds.load_crit", False), ("temp_warn", "thresholds.temp_warn", False), ("temp_crit", "thresholds.temp_crit", False), ("uptime_warn", "thresholds.uptime_warn", True), ("uptime_crit", "thresholds.uptime_crit", True), ("latency_warn", "ping.threshold.good_ms", False), ("latency_crit", "ping.threshold.warn_ms", False), ] # Docker time-series retention windows (days). Read fresh by the cleanup task, # so a save takes effect on the next hourly run — no app.config wiring needed. _RETENTION_FIELDS = [ ("docker_metrics_raw_days", "docker.retention.metrics_raw_days"), ("docker_metrics_rollup_days", "docker.retention.metrics_rollup_days"), ("docker_events_days", "docker.retention.events_days"), ("metrics_raw_days", "metrics.retention.raw_days"), ("metrics_rollup_days", "metrics.retention.rollup_days"), ] @settings_bp.get("/thresholds/") @require_role(UserRole.admin) async def thresholds(): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) return await render_template("settings/thresholds.html", settings=settings) @settings_bp.post("/thresholds/") @require_role(UserRole.admin) async def save_thresholds(): form = await request.form async with current_app.db_sessionmaker() as db: async with db.begin(): for field, key, is_float in _THRESHOLD_FIELDS: raw = form.get(field, "") if raw == "": continue try: val = float(raw) if is_float else int(raw) except (TypeError, ValueError): continue await set_setting(db, key, val) for field, key in _RETENTION_FIELDS: raw = form.get(field, "") if raw == "": continue try: val = max(1, int(raw)) # at least a day — 0 would prune everything except (TypeError, ValueError): continue await set_setting(db, key, val) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "thresholds"}) return redirect(url_for("settings.thresholds")) # ── Notifications (SMTP + Webhook) ──────────────────────────────────────────── @settings_bp.get("/notifications/") @require_role(UserRole.admin) async def notifications(): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) return await render_template("settings/notifications.html", settings=settings) @settings_bp.post("/notifications/") @require_role(UserRole.admin) async def save_notifications(): form = await request.form async with current_app.db_sessionmaker() as db: async with db.begin(): recipients_raw = form.get("smtp.recipients", "") recipients = [r.strip() for r in recipients_raw.split(",") if r.strip()] await set_setting(db, "smtp.host", form.get("smtp.host", "")) await set_setting(db, "smtp.port", int(form.get("smtp.port", 587))) await set_setting(db, "smtp.tls", "smtp.tls" in form) await set_setting(db, "smtp.username", form.get("smtp.username", "")) if form.get("smtp.password"): await set_setting(db, "smtp.password", form.get("smtp.password")) await set_setting(db, "smtp.recipients", recipients) await set_setting(db, "webhook.url", form.get("webhook.url", "")) await set_setting(db, "webhook.template", form.get("webhook.template", "")) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "notifications"}) return redirect(url_for("settings.notifications")) # ── Ansible Sources ─────────────────────────────────────────────────────────── async def _get_ansible_sources() -> list[dict]: async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) sources = settings.get("ansible.sources", []) return sources if isinstance(sources, list) else [] async def _save_ansible_sources(sources: list[dict]) -> None: async with current_app.db_sessionmaker() as db: async with db.begin(): await set_setting(db, "ansible.sources", sources) await _reload_app_config() async def _ansible_sources_partial(sources: list[dict]): return await render_template("settings/_ansible_sources.html", sources=sources) @settings_bp.get("/ansible/") @require_role(UserRole.admin) async def ansible(): sources = await _get_ansible_sources() async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) return await render_template( "settings/ansible.html", sources=sources, ssh_key_set=bool(settings.get("ansible.ssh_private_key")), become_set=bool(settings.get("ansible.become_password")), vault_set=bool(settings.get("ansible.vault_password")), ssh_public_key=settings.get("ansible.ssh_public_key", ""), ssh_user=settings.get("ansible.ssh_user", "steward"), host_key_checking=settings.get("ansible.host_key_checking", False), max_concurrent_runs=settings.get("ansible.max_concurrent_runs", 3), ) @settings_bp.post("/ansible/generate-key") @require_role(UserRole.admin) async def ansible_generate_key(): """Mint a fresh managed ed25519 keypair (private encrypted, public shown). This is the reusable Steward identity: provisioning installs the public half on every host, and steady-state runs authenticate with the private half. Regenerating invalidates access to already-provisioned hosts until they are re-provisioned, so the UI confirms before calling this. """ from steward.core.crypto import generate_ssh_keypair form = await request.form private_pem, public_line = generate_ssh_keypair() async with current_app.db_sessionmaker() as db: async with db.begin(): await set_setting(db, "ansible.ssh_private_key", private_pem) await set_setting(db, "ansible.ssh_public_key", public_line) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "ansible.managed_key"}) # Allow the inline trigger on other pages (e.g. Host Agents) to return there. nxt = (form.get("next", "") or "").strip() if nxt.startswith("/") and not nxt.startswith("//"): return redirect(nxt) return redirect(url_for("settings.ansible")) @settings_bp.post("/ansible/credentials") @require_role(UserRole.admin) async def ansible_save_credentials(): """Save global Ansible credentials (masked-update: blank keeps the current value).""" form = await request.form async with current_app.db_sessionmaker() as db: async with db.begin(): for field, key in ( ("ssh_private_key", "ansible.ssh_private_key"), ("become_password", "ansible.become_password"), ("vault_password", "ansible.vault_password"), ): if f"clear_{field}" in form: await set_setting(db, key, "") continue raw = form.get(field, "") or "" # Preserve PEM whitespace for the key; trim passwords. val = raw.strip("\n") if field == "ssh_private_key" else raw.strip() if val: await set_setting(db, key, val) ssh_user = (form.get("ssh_user", "") or "").strip() if ssh_user: await set_setting(db, "ansible.ssh_user", ssh_user) await set_setting(db, "ansible.host_key_checking", "host_key_checking" in form) try: mcr = max(1, min(50, int(form.get("max_concurrent_runs", 3)))) except (TypeError, ValueError): mcr = 3 await set_setting(db, "ansible.max_concurrent_runs", mcr) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "ansible.credentials"}) return redirect(url_for("settings.ansible")) @settings_bp.post("/ansible/sources/add") @require_role(UserRole.admin) async def ansible_add_source(): form = await request.form name = form.get("name", "").strip() src_type = form.get("type", "local").strip() if not name: sources = await _get_ansible_sources() return await _ansible_sources_partial(sources) entry: dict = {"name": name, "type": src_type} if src_type == "git": entry["url"] = form.get("url", "").strip() entry["branch"] = form.get("branch", "main").strip() or "main" try: entry["pull_interval_seconds"] = int(form.get("pull_interval_seconds", 3600)) except ValueError: entry["pull_interval_seconds"] = 3600 token = form.get("http_token", "").strip() if token: entry["http_token"] = token else: entry["path"] = form.get("path", "").strip() sources = await _get_ansible_sources() sources.append(entry) await _save_ansible_sources(sources) return await _ansible_sources_partial(sources) @settings_bp.post("/ansible/sources//remove") @require_role(UserRole.admin) async def ansible_remove_source(idx: int): sources = await _get_ansible_sources() if 0 <= idx < len(sources): sources.pop(idx) await _save_ansible_sources(sources) return await _ansible_sources_partial(sources) @settings_bp.post("/ansible/sources//save") @require_role(UserRole.admin) async def ansible_save_source(idx: int): sources = await _get_ansible_sources() if not (0 <= idx < len(sources)): return await _ansible_sources_partial(sources) form = await request.form src_type = form.get("type", "local").strip() entry: dict = { "name": form.get("name", sources[idx].get("name", "")).strip(), "type": src_type, } if src_type == "git": entry["url"] = form.get("url", "").strip() entry["branch"] = form.get("branch", "main").strip() or "main" try: entry["pull_interval_seconds"] = int(form.get("pull_interval_seconds", 3600)) except ValueError: entry["pull_interval_seconds"] = 3600 token = form.get("http_token", "").strip() if token: entry["http_token"] = token elif "http_token" in sources[idx]: # Preserve existing token — password fields don't round-trip via browser entry["http_token"] = sources[idx]["http_token"] else: entry["path"] = form.get("path", "").strip() sources[idx] = entry await _save_ansible_sources(sources) return await _ansible_sources_partial(sources) @settings_bp.post("/ansible/sources//sync") @require_role(UserRole.admin) async def ansible_sync_source(idx: int): """Trigger a git pull for a git source and return a status badge.""" sources = await _get_ansible_sources() if not (0 <= idx < len(sources)): return ('Source not found', 404) source = sources[idx] if source.get("type") != "git": return 'Not a git source' from steward.ansible.sources import git_pull from steward.core.settings import to_ansible_cfg async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) ansible_cfg = to_ansible_cfg(settings) from steward.ansible.sources import get_sources resolved = next((s for s in get_sources(ansible_cfg) if s["name"] == source["name"]), None) if resolved is None: return 'Could not resolve source path' try: await git_pull(resolved) return 'Synced' except Exception as exc: logger.exception("git pull failed for source %r", source["name"]) return f'Sync failed: {exc}' # ── Auth (OIDC / LDAP) ──────────────────────────────────────────────────────── @settings_bp.get("/auth/") @require_role(UserRole.admin) async def auth_settings(): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) return await render_template("settings/auth.html", settings=settings) @settings_bp.post("/auth/") @require_role(UserRole.admin) async def save_auth_settings(): form = await request.form async with current_app.db_sessionmaker() as db: async with db.begin(): # OIDC await set_setting(db, "oidc.enabled", "oidc.enabled" in form) await set_setting(db, "oidc.discovery_url", form.get("oidc.discovery_url", "").strip()) await set_setting(db, "oidc.client_id", form.get("oidc.client_id", "").strip()) if form.get("oidc.client_secret"): await set_setting(db, "oidc.client_secret", form.get("oidc.client_secret").strip()) await set_setting(db, "oidc.scopes", form.get("oidc.scopes", "openid profile email").strip()) await set_setting(db, "oidc.username_claim", form.get("oidc.username_claim", "preferred_username").strip()) await set_setting(db, "oidc.email_claim", form.get("oidc.email_claim", "email").strip()) await set_setting(db, "oidc.groups_claim", form.get("oidc.groups_claim", "groups").strip()) await set_setting(db, "oidc.admin_group", form.get("oidc.admin_group", "").strip()) await set_setting(db, "oidc.operator_group", form.get("oidc.operator_group", "").strip()) # LDAP await set_setting(db, "ldap.enabled", "ldap.enabled" in form) await set_setting(db, "ldap.host", form.get("ldap.host", "").strip()) await set_setting(db, "ldap.port", int(form.get("ldap.port") or 389)) await set_setting(db, "ldap.tls", "ldap.tls" in form) await set_setting(db, "ldap.bind_dn", form.get("ldap.bind_dn", "").strip()) if form.get("ldap.bind_password"): await set_setting(db, "ldap.bind_password", form.get("ldap.bind_password").strip()) await set_setting(db, "ldap.base_dn", form.get("ldap.base_dn", "").strip()) await set_setting(db, "ldap.user_filter", form.get("ldap.user_filter", "(uid={username})").strip()) await set_setting(db, "ldap.admin_group_dn", form.get("ldap.admin_group_dn", "").strip()) await set_setting(db, "ldap.operator_group_dn", form.get("ldap.operator_group_dn", "").strip()) await set_setting(db, "ldap.attr_username", form.get("ldap.attr_username", "uid").strip()) await set_setting(db, "ldap.attr_email", form.get("ldap.attr_email", "mail").strip()) await _reload_app_config() await log_audit(current_app, session.get("user_id"), session.get("username", ""), "settings.saved", detail={"section": "auth"}) # Clear OIDC discovery cache so new settings take effect from steward.auth.oidc import _discovery_cache _discovery_cache.clear() return redirect(url_for("settings.auth_settings")) # ── Reports ─────────────────────────────────────────────────────────────────── @settings_bp.get("/reports/") @require_role(UserRole.admin) async def reports(): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) return await render_template("settings/reports.html", settings=settings, flash=None) @settings_bp.post("/reports/") @require_role(UserRole.admin) async def save_reports(): form = await request.form async with current_app.db_sessionmaker() as db: async with db.begin(): await set_setting(db, "reports.enabled", "reports.enabled" in form) await set_setting(db, "reports.schedule_day", int(form.get("reports.schedule_day", 6))) await set_setting(db, "reports.schedule_hour", int(form.get("reports.schedule_hour", 8))) return redirect(url_for("settings.reports")) @settings_bp.post("/reports/send-now/") @require_role(UserRole.admin) async def send_report_now(): from steward.core.reports import send_report async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) ok, message = await send_report(current_app._get_current_object()) return await render_template( "settings/reports.html", settings=settings, flash={"ok": ok, "message": message}, ) # ── Plugin repository helpers ───────────────────────────────────────────────── def _get_plugin_repos(settings: dict) -> list[dict]: """Return the list of plugin repositories from settings. Migrates the legacy plugins.index_url single-URL field to the list format on first access if plugins.repositories is not yet set. """ repos = settings.get("plugins.repositories") if repos and isinstance(repos, list): return repos # Migrate legacy single URL legacy = settings.get("plugins.index_url", "").strip() if legacy: return [{"name": "Default", "url": legacy}] return [] async def _save_plugin_repos(db, repos: list[dict]) -> None: await set_setting(db, "plugins.repositories", repos) async def _plugin_repos_partial(settings: dict): repos = _get_plugin_repos(settings) return await render_template("settings/_plugin_repos.html", repos=repos) def _build_plugin_cfg_from_form(plugin: dict, form) -> dict: """Extract config values for one plugin from a form submission.""" name = plugin["_dir"] enabled = f"plugin.{name}.enabled" in form plugin_cfg: dict = {"enabled": enabled} for cfg_key in plugin.get("config", {}).keys(): field_name = f"plugin.{name}.{cfg_key}" default_val = plugin["config"][cfg_key] if isinstance(default_val, list): pass # list configs must be set in plugin.yaml elif isinstance(default_val, dict): sub_dict = {} for sub_key, sub_default in default_val.items(): sub_field = f"plugin.{name}.{cfg_key}.{sub_key}" if isinstance(sub_default, bool): sub_dict[sub_key] = sub_field in form elif sub_field in form: raw_sub = form[sub_field] if isinstance(sub_default, int): try: sub_dict[sub_key] = int(raw_sub) except ValueError: sub_dict[sub_key] = sub_default else: sub_dict[sub_key] = raw_sub else: sub_dict[sub_key] = sub_default plugin_cfg[cfg_key] = sub_dict elif field_name in form: raw_val = form[field_name] if isinstance(default_val, bool): plugin_cfg[cfg_key] = raw_val.lower() in ("1", "true", "yes", "on") elif isinstance(default_val, int): try: plugin_cfg[cfg_key] = int(raw_val) except ValueError: plugin_cfg[cfg_key] = default_val else: plugin_cfg[cfg_key] = raw_val return plugin_cfg # ── Plugins list ────────────────────────────────────────────────────────────── # Bundled plugins that are really built-in *host/monitoring capabilities* (facets # of a host), not discrete vendor integrations. Presentation-only split — they # still load through the normal plugin mechanism. A plugin.yaml may set # kind: capability|integration to override. CAPABILITY_PLUGINS = {"host_agent", "http", "snmp", "docker"} @settings_bp.get("/plugins/") @require_role(UserRole.admin) async def plugins(): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) discovered = _discover_plugins() _merge_plugin_config(discovered, to_plugins_cfg(settings)) for p in discovered: p["_kind"] = p.get("kind") or ( "capability" if p["_dir"] in CAPABILITY_PLUGINS else "integration") repos = _get_plugin_repos(settings) return await render_template( "settings/plugins.html", capabilities=[p for p in discovered if p["_kind"] == "capability"], integrations=[p for p in discovered if p["_kind"] != "capability"], discovered_plugins=discovered, repos=repos, settings=settings, ) # ── Per-plugin detail (settings) ────────────────────────────────────────────── @settings_bp.get("/plugins//") @require_role(UserRole.admin) async def plugin_detail(name: str): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) discovered = _discover_plugins() plugin = next((p for p in discovered if p["_dir"] == name), None) if plugin is None: return redirect(url_for("settings.plugins")) _merge_plugin_config([plugin], to_plugins_cfg(settings)) from steward.core.plugin_manager import _LOADED_PLUGINS, get_plugin_failures return await render_template( "settings/plugin_detail.html", plugin=plugin, is_loaded=name in _LOADED_PLUGINS, fail_reason=get_plugin_failures().get(name), ) @settings_bp.post("/plugins//") @require_role(UserRole.admin) async def save_plugin_detail(name: str): form = await request.form discovered = _discover_plugins() plugin = next((p for p in discovered if p["_dir"] == name), None) if plugin is None: return redirect(url_for("settings.plugins")) async with current_app.db_sessionmaker() as db: old_settings = await get_all_settings(db) old_enabled = to_plugins_cfg(old_settings).get(name, {}).get("enabled", False) plugin_cfg = _build_plugin_cfg_from_form(plugin, form) async with current_app.db_sessionmaker() as db: async with db.begin(): await set_setting(db, f"plugin.{name}", plugin_cfg) await _reload_app_config() from steward.core.plugin_index import clear_catalog_cache clear_catalog_cache() new_enabled = plugin_cfg.get("enabled", False) if old_enabled != new_enabled: action = "plugin.enabled" if new_enabled else "plugin.disabled" await log_audit(current_app, session.get("user_id"), session.get("username", ""), action, entity_type="plugin", entity_id=name) if new_enabled and not old_enabled: from steward.core.plugin_manager import hot_reload_plugin hot_reload_plugin(current_app._get_current_object(), name) return redirect(url_for("settings.plugin_detail", name=name)) # ── Plugin repository HTMX routes ───────────────────────────────────────────── @settings_bp.post("/plugins/repos/add") @require_role(UserRole.admin) async def plugin_repos_add(): form = await request.form async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) repos = _get_plugin_repos(settings) url = form.get("url", "").strip() repo_name = form.get("name", "").strip() or url if url: repos.append({"name": repo_name, "url": url}) async with current_app.db_sessionmaker() as db: async with db.begin(): await _save_plugin_repos(db, repos) from steward.core.plugin_index import clear_catalog_cache clear_catalog_cache() return await _plugin_repos_partial({"plugins.repositories": repos}) @settings_bp.post("/plugins/repos//remove") @require_role(UserRole.admin) async def plugin_repos_remove(idx: int): async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) repos = _get_plugin_repos(settings) if 0 <= idx < len(repos): repos.pop(idx) async with current_app.db_sessionmaker() as db: async with db.begin(): await _save_plugin_repos(db, repos) from steward.core.plugin_index import clear_catalog_cache clear_catalog_cache() return await _plugin_repos_partial({"plugins.repositories": repos}) @settings_bp.post("/plugins/repos//save") @require_role(UserRole.admin) async def plugin_repos_save(idx: int): form = await request.form async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) repos = _get_plugin_repos(settings) if 0 <= idx < len(repos): repos[idx] = { "name": form.get("name", "").strip() or repos[idx]["name"], "url": form.get("url", "").strip() or repos[idx]["url"], } async with current_app.db_sessionmaker() as db: async with db.begin(): await _save_plugin_repos(db, repos) from steward.core.plugin_index import clear_catalog_cache clear_catalog_cache() return await _plugin_repos_partial({"plugins.repositories": repos}) # ── Plugin catalog (HTMX partial) ───────────────────────────────────────────── @settings_bp.get("/plugins/catalog/") @require_role(UserRole.admin) async def plugins_catalog(): """HTMX partial: list plugins available in the remote catalog.""" async with current_app.db_sessionmaker() as db: settings = await get_all_settings(db) repos = _get_plugin_repos(settings) repo_urls = [r["url"] for r in repos if r.get("url")] if not repo_urls: return await render_template( "settings/_catalog_partial.html", catalog=[], installed_names=set(), loaded_names=set(), error="No plugin repositories configured.", ) from steward.core.plugin_index import fetch_catalog from steward.core.plugin_manager import _LOADED_PLUGINS force = request.args.get("refresh") == "1" try: catalog = await fetch_catalog(repo_urls, force=force) error = None if catalog else "Catalog is empty or could not be fetched." except Exception as exc: catalog = [] error = str(exc) discovered = _discover_plugins() installed_names = {p["_dir"] for p in discovered} installed_versions = {p["_dir"]: p.get("version", "0.0.0") for p in discovered} return await render_template( "settings/_catalog_partial.html", catalog=catalog, installed_names=installed_names, installed_versions=installed_versions, loaded_names=_LOADED_PLUGINS, error=error, ) # ── Plugin install ───────────────────────────────────────────────────────────── @settings_bp.post("/plugins/install/") @require_role(UserRole.admin) async def install_plugin(name: str): """Download and install a plugin from the catalog, then hot-reload it.""" form = await request.form download_url = form.get("download_url", "").strip() checksum = form.get("checksum_sha256", "").strip() if not download_url: return await render_template( "settings/_install_result.html", name=name, success=False, message="No download URL provided.", ) from steward.core.plugin_manager import download_and_install_plugin, hot_reload_plugin ok, msg = await download_and_install_plugin( current_app._get_current_object(), name, download_url, checksum ) if not ok: return await render_template( "settings/_install_result.html", name=name, success=False, message=msg, ) # Attempt hot-reload ok2, msg2 = hot_reload_plugin(current_app._get_current_object(), name) if ok2: return await render_template( "settings/_install_result.html", name=name, success=True, message=f"Installed and activated. {msg2}", ) else: return await render_template( "settings/_install_result.html", name=name, success=True, message=f"Installed. {msg2}", restart_required=True, ) # ── Plugin hot-reload ────────────────────────────────────────────────────────── @settings_bp.post("/plugins/reload/") @require_role(UserRole.admin) async def reload_plugin(name: str): """Hot-reload a plugin that was installed but not yet active.""" from steward.core.plugin_manager import hot_reload_plugin ok, msg = hot_reload_plugin(current_app._get_current_object(), name) return await render_template( "settings/_install_result.html", name=name, success=ok, message=msg, restart_required=(not ok and "restart" in msg.lower()), ) # ── App restart ─────────────────────────────────────────────────────────────── @settings_bp.post("/plugins/restart/") @require_role(UserRole.admin) async def restart_app_route(): """Trigger an in-app restart (replaces the process via os.execv).""" import asyncio from steward.core.plugin_manager import restart_app # Schedule the restart slightly after we return the response async def _delayed_restart(): await asyncio.sleep(0.5) restart_app() asyncio.create_task(_delayed_restart()) return await render_template("settings/_restart_pending.html")