bvandeusenandClaude Opus 5.5
8941ed0372
Migration 0030 seeded plugin_metric_latest from the whole history: two index probes per distinct metric key. On the live instance that held the first boot for 7+ minutes with Steward down. The migration now only creates the table. Live sources fill it on their next sample, and backfill_latest runs after startup for resources that have history but no latest rows. Per resource it does one backward probe for the last sample, then takes the newest value per metric within the hour before it. Resources that already have rows are skipped, so later boots only walk the resource names. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
501 lines
22 KiB
Python
501 lines
22 KiB
Python
# steward/app.py
|
|
from __future__ import annotations
|
|
import asyncio
|
|
import logging
|
|
from pathlib import Path
|
|
from quart import Quart, g, render_template, request
|
|
from .config import load_bootstrap
|
|
from .database import (
|
|
init_db, ensure_database_reachable, DatabaseUnavailable,
|
|
DB_CONNECT_TIMEOUT_SECONDS,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def create_app(
|
|
config_path: Path | str | None = None,
|
|
testing: bool = False,
|
|
) -> Quart:
|
|
app = Quart(__name__, template_folder="templates")
|
|
|
|
# ── 1. Bootstrap: db_url + secret_key only ────────────────────────────────
|
|
if not testing:
|
|
bootstrap = load_bootstrap(config_path)
|
|
else:
|
|
bootstrap = {
|
|
"database_url": "postgresql+asyncpg://test/test",
|
|
"secret_key": "test-secret-key",
|
|
"db_connect_timeout": DB_CONNECT_TIMEOUT_SECONDS,
|
|
"plugin_dirs": ["plugins"],
|
|
"plugin_install_dir": "plugins",
|
|
}
|
|
|
|
app.config.update(
|
|
SECRET_KEY=bootstrap["secret_key"],
|
|
DATABASE_URL=bootstrap["database_url"],
|
|
PLUGIN_DIRS=bootstrap["plugin_dirs"],
|
|
PLUGIN_INSTALL_DIR=bootstrap["plugin_install_dir"],
|
|
TESTING=testing,
|
|
# Lax keeps the session cookie off cross-site POSTs and iframes; the
|
|
# CSRF token below is the check that does not depend on the browser.
|
|
SESSION_COOKIE_SAMESITE="Lax",
|
|
)
|
|
|
|
# ── 2. Init DB engine ──────────────────────────────────────────────────────
|
|
if not testing:
|
|
init_db(app)
|
|
else:
|
|
from unittest.mock import MagicMock
|
|
app.db_sessionmaker = MagicMock()
|
|
|
|
# ── 2b. Block until the database is actually reachable ────────────────────
|
|
# Everything from here down touches the DB synchronously (migrations,
|
|
# secret re-encryption, settings load). Gate all of it behind one readiness
|
|
# check so a DB that is merely slow to come up — WAL recovery after an
|
|
# unclean shutdown, or Docker DNS not yet serving the `db` name after a host
|
|
# reboot — is waited out instead of crashing the container on a traceback.
|
|
if not testing:
|
|
from .database import wait_for_database
|
|
wait_for_database(
|
|
app.config["DATABASE_URL"],
|
|
timeout_seconds=bootstrap["db_connect_timeout"],
|
|
)
|
|
|
|
# ── 3. Core migrations only (creates app_settings table) ──────────────────
|
|
if not testing:
|
|
from .core.migration_runner import run_core_migrations
|
|
run_core_migrations(
|
|
app.config["DATABASE_URL"],
|
|
plugin_dirs=[Path(d).resolve() for d in app.config["PLUGIN_DIRS"]],
|
|
)
|
|
|
|
# ── 3b. Encryption-at-rest: bind the encryptor, then encrypt legacy secrets ─
|
|
if not testing:
|
|
from .core.crypto import init_crypto
|
|
from .core.settings import migrate_plaintext_secrets, scan_undecryptable_secrets
|
|
init_crypto(app.config["SECRET_KEY"])
|
|
migrate_plaintext_secrets(app.config["DATABASE_URL"])
|
|
# Flag any secrets sealed under a now-lost key (see the admin banner).
|
|
undecryptable = scan_undecryptable_secrets(app.config["DATABASE_URL"])
|
|
if undecryptable:
|
|
logger.warning(
|
|
"%d stored secret(s) cannot be decrypted with the current app key "
|
|
"(re-enter them in Settings): %s",
|
|
len(undecryptable), ", ".join(undecryptable),
|
|
)
|
|
|
|
# ── 4. Load all settings from DB → populate app.config ────────────────────
|
|
if not testing:
|
|
from .core.settings import (
|
|
load_settings_sync, to_smtp_cfg, to_webhook_cfg,
|
|
to_ansible_cfg, to_plugins_cfg, to_oidc_cfg, to_ldap_cfg,
|
|
to_thresholds_cfg, to_access_cfg,
|
|
)
|
|
_settings = load_settings_sync(app.config["DATABASE_URL"])
|
|
from .core.logging_setup import set_log_level
|
|
set_log_level(_settings.get("log.level"))
|
|
app.config.update(to_access_cfg(_settings))
|
|
app.config.update(
|
|
DATA_RETENTION_DAYS=_settings.get("data.retention_days", 90),
|
|
MONITORS_POLL_INTERVAL=_settings.get("monitors.poll_interval_seconds", 60),
|
|
PUBLIC_BASE_URL=_settings.get("general.public_base_url", ""),
|
|
SMTP=to_smtp_cfg(_settings),
|
|
WEBHOOK=to_webhook_cfg(_settings),
|
|
ANSIBLE=to_ansible_cfg(_settings),
|
|
PLUGINS=to_plugins_cfg(_settings),
|
|
OIDC=to_oidc_cfg(_settings),
|
|
LDAP=to_ldap_cfg(_settings),
|
|
THRESHOLDS=to_thresholds_cfg(_settings),
|
|
)
|
|
else:
|
|
from .core.settings import to_access_cfg, to_thresholds_cfg
|
|
app.config.update(to_access_cfg({}))
|
|
app.config.update(
|
|
DATA_RETENTION_DAYS=90,
|
|
MONITORS_POLL_INTERVAL=60,
|
|
PUBLIC_BASE_URL="",
|
|
SMTP={},
|
|
WEBHOOK={},
|
|
ANSIBLE={},
|
|
PLUGINS={},
|
|
OIDC={"enabled": False},
|
|
LDAP={"enabled": False},
|
|
THRESHOLDS=to_thresholds_cfg({}),
|
|
)
|
|
|
|
# `threshold_style(value, kind)` — degraded-value coloring for templates,
|
|
# reading the configurable cutoffs from app.config["THRESHOLDS"]. Available
|
|
# in every template (core + plugin) via the jinja global.
|
|
from .core.settings import threshold_style_for as _ts_for
|
|
app.jinja_env.globals["threshold_style"] = (
|
|
lambda value, kind: _ts_for(value, kind, app.config.get("THRESHOLDS", {}))
|
|
)
|
|
|
|
# One way to write a size, a rate, a span and a moment, and to grade a
|
|
# reading ('ok' / 'warn' / 'crit' / 'unknown') against the same cutoffs.
|
|
from .core import formatting as _fmt
|
|
app.jinja_env.filters["bytes"] = _fmt.human_bytes
|
|
app.jinja_env.filters["rate"] = _fmt.human_rate
|
|
app.jinja_env.filters["duration"] = _fmt.human_duration
|
|
app.jinja_env.filters["ago"] = _fmt.ago
|
|
app.jinja_env.globals["status_class"] = (
|
|
lambda value, kind: _fmt.status_class(value, kind, app.config.get("THRESHOLDS", {}))
|
|
)
|
|
# The status and layout macros in _macros.html read these.
|
|
app.jinja_env.globals["status_info"] = _fmt.status_info
|
|
app.jinja_env.globals["required"] = _fmt.required
|
|
from .models.monitors import MONITOR_TYPE_LABELS as _type_labels
|
|
app.jinja_env.filters["monitor_type"] = lambda t: _type_labels.get(t, str(t).upper())
|
|
|
|
# Forged-request protection: every POST form renders csrf_field(), HTMX and
|
|
# fetch() send csrf_token() as X-CSRF-Token, and one check refuses any
|
|
# unsafe request without it (steward/auth/csrf.py).
|
|
from .auth import csrf as _csrf
|
|
app.jinja_env.globals["csrf_token"] = _csrf.csrf_token
|
|
app.jinja_env.globals["csrf_field"] = _csrf.csrf_field
|
|
app.before_request(_csrf.check_request)
|
|
# The session cookie is Secure, and HSTS is sent, only on requests that
|
|
# arrived over HTTPS — directly or as a trusted proxy reports it
|
|
# (steward/core/proxy.py). Plain-HTTP and LAN use keep working.
|
|
from quart import has_request_context
|
|
from quart.sessions import SecureCookieSessionInterface
|
|
from .core.proxy import is_https
|
|
|
|
class _SessionInterface(SecureCookieSessionInterface):
|
|
def get_cookie_secure(self, app):
|
|
return bool(app.config.get("SESSION_COOKIE_SECURE")) or (
|
|
has_request_context() and is_https())
|
|
|
|
app.session_interface = _SessionInterface()
|
|
|
|
@app.after_request
|
|
async def _hsts(response):
|
|
if is_https():
|
|
response.headers.setdefault("Strict-Transport-Security", "max-age=31536000")
|
|
return response
|
|
|
|
# Default-deny: a view with neither require_role nor public_route is refused.
|
|
from .auth.middleware import check_view_is_marked, public_route
|
|
app.before_request(check_view_is_marked)
|
|
|
|
# ── 5. Plugin migrations ───────────────────────────────────────────────────
|
|
# Step 3 already applied all plugin migrations via discover_all_plugin_migration_dirs,
|
|
# so this is a no-op on normal startup. We still run it with all discovered dirs so
|
|
# Alembic can resolve the full revision graph regardless of which plugins are enabled.
|
|
if not testing:
|
|
from .core.migration_runner import run_plugin_migrations, discover_all_in
|
|
_plugin_dirs = [Path(d).resolve() for d in app.config["PLUGIN_DIRS"]]
|
|
_all_plugin_dirs = discover_all_in(_plugin_dirs)
|
|
run_plugin_migrations(app.config["DATABASE_URL"], _all_plugin_dirs)
|
|
|
|
# ── 6. Alert pipeline ──────────────────────────────────────────────────────
|
|
from .core.alerts import init_alerts
|
|
init_alerts(app)
|
|
|
|
# ── 7. Register core blueprints ────────────────────────────────────────────
|
|
from .auth.routes import auth_bp
|
|
from .dashboard.routes import dashboard_bp
|
|
from .hosts.routes import hosts_bp
|
|
from .monitors.routes import monitors_bp
|
|
from .status.routes import status_bp
|
|
from .alerts.routes import alerts_bp
|
|
from .ansible.routes import ansible_bp
|
|
from .ansible.inventory_routes import inventory_bp
|
|
from .settings.routes import settings_bp
|
|
from .audit.routes import audit_bp
|
|
from .ai_access.routes import ai_access_bp
|
|
|
|
app.register_blueprint(auth_bp)
|
|
app.register_blueprint(dashboard_bp)
|
|
app.register_blueprint(hosts_bp)
|
|
app.register_blueprint(monitors_bp)
|
|
app.register_blueprint(status_bp)
|
|
app.register_blueprint(alerts_bp)
|
|
app.register_blueprint(ansible_bp)
|
|
app.register_blueprint(inventory_bp)
|
|
app.register_blueprint(settings_bp)
|
|
app.register_blueprint(audit_bp)
|
|
app.register_blueprint(ai_access_bp)
|
|
|
|
# Register the unified Monitor status source for the Status page. Plugins
|
|
# may register additional sources from setup(). Idempotent.
|
|
from .core.status import register_status_source, monitor_status_source
|
|
register_status_source(monitor_status_source)
|
|
|
|
# Publish the Ansible "run a playbook" capability so plugins (e.g. host_agent
|
|
# auto-deploy) can drive runs without importing the runner. Ansible is core,
|
|
# so this is always available; consumers still gate on has_capability().
|
|
from .core.capabilities import register_capability
|
|
from .ansible.runner import trigger_run
|
|
from .models.users import UserRole as _UserRole
|
|
register_capability(
|
|
"ansible.run_playbook", trigger_run,
|
|
label="Run Ansible playbook",
|
|
description="Launch an Ansible playbook run (manual, alert, schedule, or plugin-driven).",
|
|
required_role=_UserRole.operator,
|
|
)
|
|
|
|
# ── 8. Build task registry ─────────────────────────────────────────────────
|
|
app._task_registry = []
|
|
|
|
if not testing:
|
|
_register_core_tasks(app)
|
|
|
|
# ── 9. Load plugins ────────────────────────────────────────────────────────
|
|
if not testing:
|
|
from .core.plugin_manager import load_plugins
|
|
load_plugins(app)
|
|
|
|
# ── 10. Template context: inject plugin_failures into every response ───────
|
|
@app.context_processor
|
|
def _inject_plugin_failures():
|
|
from .core.plugin_manager import get_plugin_failures, get_plugin_nav
|
|
from .core.settings import get_undecryptable_secrets
|
|
# enabled_plugins lets templates gate cross-plugin embeds (e.g. the
|
|
# Hosts hub only fetches the docker fragment when docker is enabled),
|
|
# avoiding a 404 to a route whose blueprint isn't registered.
|
|
enabled_plugins = {
|
|
name for name, cfg in (app.config.get("PLUGINS") or {}).items()
|
|
if isinstance(cfg, dict) and cfg.get("enabled")
|
|
}
|
|
# Sidebar entries for enabled plugins that expose a UI (get_nav). Filter
|
|
# by enabled so a hot-disabled plugin's link can't linger before restart.
|
|
plugin_nav = [e for e in get_plugin_nav() if e["plugin"] in enabled_plugins]
|
|
return {
|
|
"plugin_failures": get_plugin_failures(),
|
|
"enabled_plugins": enabled_plugins,
|
|
"plugin_nav": plugin_nav,
|
|
# Read from an in-memory cache (no DB hit per render); kept current by
|
|
# the startup scan + set_setting's discard-on-write.
|
|
"undecryptable_secrets": get_undecryptable_secrets(),
|
|
}
|
|
|
|
@app.context_processor
|
|
async def _inject_alert_count():
|
|
"""The sidebar's count of alerts needing attention. Full pages only:
|
|
an htmx fragment has no sidebar, and a logged-out page has no nav."""
|
|
from quart import session
|
|
if (request.headers.get("HX-Request") or not session.get("user_id")
|
|
or g.get("database_down")):
|
|
return {}
|
|
try:
|
|
from .alerts.service import active_count
|
|
async with app.db_sessionmaker() as db:
|
|
return {"active_alert_count": await active_count(db)}
|
|
except Exception:
|
|
# A count is decoration; the page renders without it.
|
|
logging.getLogger(__name__).exception("Active alert count failed")
|
|
return {}
|
|
|
|
# ── 11. Share-token middleware ─────────────────────────────────────────────
|
|
@app.before_request
|
|
async def _validate_share_token():
|
|
"""A share link's token (?s=) opens, read-only, exactly the widget
|
|
fragments its own dashboard loads (steward/dashboard/share.py)."""
|
|
from quart import request, g
|
|
raw = request.args.get("s")
|
|
if not raw or request.method != "GET":
|
|
return
|
|
try:
|
|
from .dashboard.share import share_grants
|
|
async with app.db_sessionmaker() as db:
|
|
if await share_grants(db, raw, request.path, request.args):
|
|
g.share_viewer = True
|
|
except Exception:
|
|
# A failed check grants nothing; the request falls through to login.
|
|
logging.getLogger(__name__).exception("Share token check failed")
|
|
|
|
# ── 11. Health check ───────────────────────────────────────────────────────
|
|
@app.get("/health")
|
|
@public_route("liveness probe for the container")
|
|
async def health():
|
|
return {"status": "ok"}
|
|
|
|
# ── 11b. Database availability gate ────────────────────────────────────────
|
|
# Every page in Steward reads the database, so a database that has gone away
|
|
# under a running app turns each request into an opaque 500 (this is how a
|
|
# login attempt surfaced a bare gaierror). Acquire a connection up front,
|
|
# retrying briefly to ride out a restart, and answer honestly if it stays
|
|
# down rather than failing deep inside a handler with a generic error.
|
|
if not testing:
|
|
@app.before_request
|
|
async def _database_gate():
|
|
# /health is a liveness probe for the container itself — it must
|
|
# stay answerable while the database is down, or a restart loop
|
|
# gets triggered by a dependency outage. Static files need no DB.
|
|
if request.endpoint in ("health", "static"):
|
|
return None
|
|
try:
|
|
await ensure_database_reachable(app._db_engine)
|
|
except DatabaseUnavailable as exc:
|
|
logger.error("Database unreachable while serving %s: %s",
|
|
request.path, exc)
|
|
g.database_down = True
|
|
return await render_template(
|
|
"errors/database_unavailable.html"), 503
|
|
return None
|
|
|
|
# ── 12. Error handlers ─────────────────────────────────────────────────────
|
|
@app.errorhandler(404)
|
|
async def not_found(_):
|
|
return await render_template("errors/404.html"), 404
|
|
|
|
@app.errorhandler(500)
|
|
async def server_error(_):
|
|
return await render_template("errors/500.html"), 500
|
|
|
|
# ── Session freshness: lifetime, deactivation and role changes apply per request
|
|
if not testing:
|
|
from .auth.middleware import refresh_session_user
|
|
app.before_request(refresh_session_user)
|
|
|
|
# ── MCP endpoint (/mcp) ───────────────────────────────────────────────────
|
|
# Mounted after plugins load so their register_mcp_tools() hooks are seen.
|
|
# It wraps app.asgi_app: /mcp never enters Quart's pipeline (see mcp/server.py).
|
|
from .mcp import mount_mcp
|
|
mount_mcp(app)
|
|
|
|
# ── 11. Startup hook ───────────────────────────────────────────────────────
|
|
@app.before_serving
|
|
async def startup():
|
|
if not testing:
|
|
await _mark_interrupted_runs(app)
|
|
asyncio.create_task(_run_scheduler(app))
|
|
app._latest_backfill = asyncio.create_task(_backfill_latest(app))
|
|
|
|
return app
|
|
|
|
|
|
async def _backfill_latest(app: Quart) -> None:
|
|
from .core.metric_series import backfill_latest
|
|
try:
|
|
await backfill_latest(app.db_sessionmaker, app.logger)
|
|
except Exception:
|
|
app.logger.exception("latest metrics: backfill failed; quiet resources fill on their next sample")
|
|
|
|
|
|
def _register_core_tasks(app: Quart) -> None:
|
|
from .core.scheduler import ScheduledTask
|
|
|
|
from .monitors.scheduler import DISPATCH_TICK_SECONDS as MONITOR_DISPATCH_TICK
|
|
cleanup_interval = 3600 # hourly
|
|
|
|
async def run_monitor_checks():
|
|
from .monitors.scheduler import run_due_monitors
|
|
await run_due_monitors(app)
|
|
|
|
async def run_cleanup():
|
|
from .core.cleanup import run_cleanup as _cleanup
|
|
await _cleanup(app)
|
|
|
|
async def run_alert_reconcile():
|
|
from .core.alerts import check_silence, notify_after_maintenance
|
|
for step in (check_silence, notify_after_maintenance):
|
|
try:
|
|
await step(app)
|
|
except Exception:
|
|
app.logger.exception("alert_reconcile: %s failed", step.__name__)
|
|
|
|
async def run_report_check():
|
|
from .core.reports import check_and_send
|
|
await check_and_send(app)
|
|
|
|
async def run_self_monitor():
|
|
from .core.self_monitor import record_self_metrics
|
|
await record_self_metrics(app)
|
|
|
|
app._task_registry.extend([
|
|
ScheduledTask(
|
|
name="weekly_report_check",
|
|
coro_factory=run_report_check,
|
|
interval_seconds=3600,
|
|
run_on_startup=False,
|
|
),
|
|
# Watch our own fd usage so a descriptor leak surfaces as a metric/alert
|
|
# instead of a silent Errno 24 lockup. Cheap; plugin_metrics roll up
|
|
# hourly so the 60s cadence is storage-bounded.
|
|
ScheduledTask(
|
|
name="self_monitor",
|
|
coro_factory=run_self_monitor,
|
|
interval_seconds=60,
|
|
run_on_startup=True,
|
|
),
|
|
# Rules whose series has gone quiet get a visible no_data mark; alerts
|
|
# a maintenance window muted notify once it ends.
|
|
ScheduledTask(
|
|
name="alert_reconcile",
|
|
coro_factory=run_alert_reconcile,
|
|
interval_seconds=60,
|
|
run_on_startup=False,
|
|
),
|
|
# Ticks often and only starts checks: each monitor decides whether it
|
|
# is due from its own interval (steward/monitors/scheduler.py).
|
|
ScheduledTask(
|
|
name="monitor_check",
|
|
coro_factory=run_monitor_checks,
|
|
interval_seconds=MONITOR_DISPATCH_TICK,
|
|
run_on_startup=True,
|
|
),
|
|
ScheduledTask(
|
|
name="data_cleanup",
|
|
coro_factory=run_cleanup,
|
|
interval_seconds=cleanup_interval,
|
|
run_on_startup=False,
|
|
),
|
|
])
|
|
|
|
# Pull Ansible git sources as each falls due. One task reads the sources
|
|
# as they are on every tick, so one added in Settings is pulled within
|
|
# half a minute, no restart (#5157).
|
|
async def run_git_pulls():
|
|
from .ansible.sources import pull_due_sources
|
|
await pull_due_sources(app.config.get("ANSIBLE", {}))
|
|
|
|
app._task_registry.append(
|
|
ScheduledTask(
|
|
name="ansible_git_pulls",
|
|
coro_factory=run_git_pulls,
|
|
interval_seconds=30,
|
|
run_on_startup=True,
|
|
)
|
|
)
|
|
|
|
# Fire due Ansible schedules (recurring playbook runs). Checks every minute;
|
|
# each schedule's own interval gates whether it actually runs.
|
|
async def run_ansible_schedules():
|
|
from .ansible.scheduler import run_due_schedules
|
|
await run_due_schedules(app)
|
|
|
|
app._task_registry.append(
|
|
ScheduledTask(
|
|
name="ansible_scheduled_runs",
|
|
coro_factory=run_ansible_schedules,
|
|
interval_seconds=60,
|
|
run_on_startup=False,
|
|
)
|
|
)
|
|
|
|
|
|
async def _mark_interrupted_runs(app: Quart) -> None:
|
|
from sqlalchemy import update
|
|
from .models.ansible import AnsibleRun, AnsibleRunStatus
|
|
from datetime import datetime, timezone
|
|
|
|
async with app.db_sessionmaker() as session:
|
|
async with session.begin():
|
|
await session.execute(
|
|
update(AnsibleRun)
|
|
.where(AnsibleRun.status.in_(
|
|
[AnsibleRunStatus.running, AnsibleRunStatus.queued]))
|
|
.values(
|
|
status=AnsibleRunStatus.interrupted,
|
|
finished_at=datetime.now(timezone.utc),
|
|
)
|
|
)
|
|
|
|
|
|
async def _run_scheduler(app: Quart) -> None:
|
|
from .core.scheduler import start_scheduler
|
|
await start_scheduler(app._task_registry)
|