bvandeusenandClaude Opus 5.5
edcfa4d3db
- git ships in the image; checkouts live under /data/ansible/sources (STEWARD_ANSIBLE_SOURCE_DIR), on the persistent volume. - One task, ansible_git_pulls, reads the sources every 30 s and pulls each as its interval passes: a source added in Settings is pulled without a restart, a removed one is forgotten. - git_pull: a 120 s timeout ending the process group, GIT_TERMINAL_PROMPT=0, pull --ff-only, a failed clone removed; it raises GitPullError with git's message, so the Sync button stops saying Synced after a failure (and the message is now escaped). - An alert action's vars and the firing's steward_alert_* vars reach ansible-playbook as one JSON extra-vars file, not -e key=value, which Ansible split on whitespace (a rule named 'High CPU on web'). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
490 lines
22 KiB
Python
490 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"])
|
|
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))
|
|
|
|
return app
|
|
|
|
|
|
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)
|