FabledSteward/steward/app.py

# 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"],
        }

    app.config.update(
        SECRET_KEY=bootstrap["secret_key"],
        DATABASE_URL=bootstrap["database_url"],
        PLUGIN_DIRS=bootstrap["plugin_dirs"],
        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", {}))
    )
    # `grade_style(grade)` — the same styling for a grade worked out already
    # (one graded against its own normal, #5742).
    from .core.settings import grade_style as _grade_style
    app.jinja_env.globals["grade_style"] = _grade_style

    # 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
    # A moment is written in the reader's zone, saved on their account and held
    # in the session (milestone 490); outside a request (an email, a report)
    # it is UTC.
    def _viewer_zone():
        from quart import has_request_context, session
        return _fmt.zone(session.get("tz")) if has_request_context() else _fmt.zone(None)
    app.jinja_env.filters["ago"] = lambda moment, now=None: _fmt.ago(moment, now, tz=_viewer_zone())
    app.jinja_env.filters["when"] = lambda moment: _fmt.when(moment, _viewer_zone())
    app.jinja_env.globals["tz_label"] = lambda: _fmt.tz_label(_viewer_zone())
    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
    # A metric and its source in words, a reading in its unit (alerts pages).
    for _name in ("metric_label", "metric_about", "metric_unit", "metric_value", "source_label",
                  "cert_left", "field_label"):
        app.jinja_env.globals[_name] = getattr(_fmt, _name)
    # Which plugin settings fields are credentials (a password box, never echoed).
    from .core.settings import is_plugin_secret as _is_plugin_secret
    app.jinja_env.globals["is_plugin_secret"] = _is_plugin_secret
    from .alerts.service import rule_condition as _rule_condition
    app.jinja_env.globals["rule_condition"] = _rule_condition
    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
    # The sign-in page offers "Forgot your password?" only while mail can send (milestone 497).
    app.jinja_env.globals["reset_by_email"] = lambda: bool(app.config.get("SMTP", {}).get("host"))
    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 .settings.users import users_bp
    from .audit.routes import audit_bp
    from .ai_access.routes import ai_access_bp
    from .account.routes import account_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(users_bp)
    app.register_blueprint(audit_bp)
    app.register_blueprint(ai_access_bp)
    app.register_blueprint(account_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))
            app._series_index = asyncio.create_task(_ensure_series_index(app))

    return app


async def _ensure_series_index(app: Quart) -> None:
    from .core.metric_index import ensure_series_index
    try:
        await ensure_series_index(app._db_engine, app.logger)
    except Exception:
        # Reads still work on the plain index; the next boot tries again.
        app.logger.exception("metric index: build failed; series reads use the plain index")


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
        from .core.self_monitor import record_cleanup_secs
        done = await _cleanup(app)
        # How long it took, as Steward's own metric (#5467); a failed run
        # raises past this and records nothing.
        await record_cleanup_secs(app, done["secs"])

    async def run_alert_reconcile():
        from .alerts.service import ensure_rule_states, reconcile_covering_rules
        from .core.alerts import check_silence, notify_after_maintenance
        for step in (reconcile_covering_rules, ensure_rule_states, 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)

    async def run_backup_tick():
        from .backup.job import tick
        await tick(app)

    # Monitor response times learn their normal (#5742); plugins register
    # their own metrics from setup().
    from .monitors.service import learn_response_times
    learn_response_times(app)

    async def run_metric_norms():
        from .core.norms import run_norms
        await run_norms(app)

    async def run_metric_anomalies():
        from .core.anomalies import run_anomalies
        await run_anomalies(app)

    async def run_email_outbox():
        from .core.outbox import send_due
        await send_due(app)

    from .backup.job import TICK_SECONDS as BACKUP_TICK_S
    from .core.outbox import SENDER_TICK_S
    app._task_registry.extend([
        # Steward's own daily backup to its bucket (#5673): a tick that
        # records the newest backup's age and makes one when it is due.
        ScheduledTask(
            name="backup",
            coro_factory=run_backup_tick,
            interval_seconds=BACKUP_TICK_S,
            run_on_startup=False,
        ),
        # Every email is a row until the mail server accepts it (#5506); this
        # sends what is due and retries what failed. On startup too, so what a
        # restart left unsent goes out at once.
        ScheduledTask(
            name="email_outbox",
            coro_factory=run_email_outbox,
            interval_seconds=SENDER_TICK_S,
            run_on_startup=True,
        ),
        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,
        ),
        # Covering rules gain a child for each newly covered host (#5464);
        # 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,
        ),
        # Each learned metric's normal per hour (#5739). Hourly, since a slot
        # gains one hour of history an hour; the table outlives a restart, so
        # there is nothing to rebuild at startup.
        ScheduledTask(
            name="metric_norms",
            coro_factory=run_metric_norms,
            interval_seconds=3600,
            run_on_startup=False,
        ),
        # Each learned metric's last 15 minutes against its normal (#5740):
        # opens, extends and ends anomalies, and lapses unreviewed ones.
        ScheduledTask(
            name="metric_anomalies",
            coro_factory=run_metric_anomalies,
            interval_seconds=60,
            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)