bvandeusen/FabledSteward
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8558d15eeb4d43b98cb2d1f6e4e4f3aba9cf7252
FabledSteward/fabledscryer/settings/routes.py
T
bvandeusen 8558d15eeb feat: plugin fault isolation notices + NUT managed mode fixes 
Plugin failures:
- Track load-time failures in _FAILED_PLUGINS dict with human-readable reasons
- Inject plugin_failures into all templates via context processor
- Show admin-only warning banner in base.html when any plugin fails to load
- Show inline failure notice per plugin card in Settings → Plugins

NUT managed mode:
- entrypoint.sh: make nut_setup.py failure non-fatal so the container starts
  even if nut_ups_host isn't set yet (allows fixing config via UI)
- save_plugins: when nut_managed is enabled, auto-set nut_host=localhost so
  the plugin connects to the managed NUT instance without extra manual config

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-03-23 18:04:53 -04:00

588 lines
26 KiB
Python
Raw Blame History

# fabledscryer/settings/routes.py
from __future__ import annotations
import json
import logging
from pathlib import Path
from quart import Blueprint, current_app, render_template, request, redirect, url_for, session
from fabledscryer.auth.middleware import require_role
from fabledscryer.core.audit import log_audit
from fabledscryer.models.users import UserRole
from fabledscryer.core.settings import (
DEFAULTS, get_all_settings, set_setting,
to_smtp_cfg, to_webhook_cfg, to_ansible_cfg, to_plugins_cfg,
to_oidc_cfg, to_ldap_cfg,
)
settings_bp = Blueprint("settings", __name__, url_prefix="/settings")
logger = logging.getLogger(__name__)
def _discover_plugins() -> list[dict]:
"""Scan PLUGIN_DIR for plugin.yaml files."""
import yaml
plugin_dir = Path(current_app.config.get("PLUGIN_DIR", "plugins")).resolve()
plugins = []
if not plugin_dir.exists():
return plugins
for entry in sorted(plugin_dir.iterdir()):
yaml_path = entry / "plugin.yaml"
if entry.is_dir() and yaml_path.exists():
try:
with yaml_path.open() as f:
meta = yaml.safe_load(f) or {}
meta["_dir"] = entry.name
plugins.append(meta)
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),
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),
)
# ── 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, "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"))
# ── 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()
return await render_template("settings/ansible.html", sources=sources)
@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
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/<int:idx>/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/<int:idx>/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
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/<int:idx>/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 ('<span style="color:var(--red);font-size:0.8rem;">Source not found</span>', 404)
source = sources[idx]
if source.get("type") != "git":
return '<span style="color:var(--text-muted);font-size:0.8rem;">Not a git source</span>'
from fabledscryer.ansible.sources import git_pull
from fabledscryer.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 fabledscryer.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 '<span style="color:var(--red);font-size:0.8rem;">Could not resolve source path</span>'
try:
await git_pull(resolved)
return '<span style="color:var(--green);font-size:0.8rem;">Synced</span>'
except Exception as exc:
logger.exception("git pull failed for source %r", source["name"])
return f'<span style="color:var(--red);font-size:0.8rem;">Sync failed: {exc}</span>'
# ── 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 fabledscryer.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 fabledscryer.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},
)
# ── Plugins ───────────────────────────────────────────────────────────────────
@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))
return await render_template(
"settings/plugins.html",
discovered_plugins=discovered,
settings=settings,
)
@settings_bp.post("/plugins/")
@require_role(UserRole.admin)
async def save_plugins():
form = await request.form
discovered = _discover_plugins()
# Snapshot old enabled states for audit diff
async with current_app.db_sessionmaker() as db:
old_settings = await get_all_settings(db)
old_plugins_cfg = to_plugins_cfg(old_settings)
async with current_app.db_sessionmaker() as db:
async with db.begin():
# Save plugin index URL if provided
index_url = form.get("plugins.index_url", "").strip()
await set_setting(db, "plugins.index_url", index_url)
for plugin in discovered:
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 (e.g. SNMP devices) 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
# Managed NUT: when nut_managed is on, the NUT daemon runs in this
# container — force nut_host/port to localhost so the plugin
# connects to it automatically without manual config.
if name == "ups" and plugin_cfg.get("nut_managed"):
plugin_cfg["nut_host"] = "localhost"
plugin_cfg["nut_port"] = 3493
await set_setting(db, f"plugin.{name}", plugin_cfg)
await _reload_app_config()
from fabledscryer.core.plugin_index import clear_catalog_cache
clear_catalog_cache()
# Audit plugin enable/disable changes + auto-hot-reload newly enabled plugins
from fabledscryer.core.plugin_manager import hot_reload_plugin
for plugin in discovered:
name = plugin["_dir"]
old_enabled = old_plugins_cfg.get(name, {}).get("enabled", False)
new_enabled = f"plugin.{name}.enabled" in form
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:
# Attempt to activate without requiring a manual reload click
hot_reload_plugin(current_app._get_current_object(), name)
# Write /data/nut_managed.json so entrypoint.sh can start NUT on next restart
_sync_nut_managed_config(form)
return redirect(url_for("settings.plugins"))
def _sync_nut_managed_config(form) -> None:
"""Write or remove /data/nut_managed.json based on UPS plugin managed NUT settings."""
import json
nut_managed_path = Path("/data/nut_managed.json")
nut_managed = "plugin.ups.nut_managed" in form
ups_host = form.get("plugin.ups.nut_ups_host", "").strip()
# Only write the file when managed mode is on AND a host is configured.
# Without a host, NUT can't start and the container would loop on errors.
if nut_managed and ups_host:
cfg = {
"ups_host": ups_host,
"driver": form.get("plugin.ups.nut_driver", "snmp-ups").strip(),
"ups_name": form.get("plugin.ups.ups_name", "ups").strip(),
"snmp_community": form.get("plugin.ups.nut_snmp_community", "public").strip(),
"snmp_version": form.get("plugin.ups.nut_snmp_version", "v1").strip(),
"username": form.get("plugin.ups.nut_username", "").strip(),
"password": form.get("plugin.ups.nut_password", "").strip(),
}
try:
nut_managed_path.write_text(json.dumps(cfg, indent=2))
except OSError as exc:
logger.warning("Could not write %s: %s", nut_managed_path, exc)
else:
try:
nut_managed_path.unlink(missing_ok=True)
except OSError as exc:
logger.warning("Could not remove %s: %s", nut_managed_path, exc)
# ── 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)
index_url = settings.get("plugins.index_url", "").strip()
if not index_url:
return await render_template(
"settings/_catalog_partial.html",
catalog=[],
installed_names=set(),
loaded_names=set(),
error="No plugin index URL configured.",
)
from fabledscryer.core.plugin_index import fetch_catalog
from fabledscryer.core.plugin_manager import _LOADED_PLUGINS
force = request.args.get("refresh") == "1"
try:
catalog = await fetch_catalog(index_url, 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/<name>")
@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 fabledscryer.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/<name>")
@require_role(UserRole.admin)
async def reload_plugin(name: str):
"""Hot-reload a plugin that was installed but not yet active."""
from fabledscryer.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 fabledscryer.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")
Reference in New Issue View Git Blame Copy Permalink