FabledSteward/steward/core/settings.py

# steward/core/settings.py
"""DB-backed application settings.

Keys use dotted notation (e.g. "smtp.host").
Values are JSON-encoded in the DB.

Usage in create_app() (before event loop):
    from steward.core.settings import load_settings_sync
    settings = load_settings_sync(db_url)

Usage at runtime (inside async handlers):
    from steward.core.settings import get_setting, set_setting
    async with app.db_sessionmaker() as session:
        value = await get_setting(session, "smtp.host")
        await set_setting(session, "smtp.host", "mail.example.com")
"""
from __future__ import annotations
import asyncio
import json
import logging
from collections.abc import Iterable, Mapping
from datetime import datetime, timezone
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from steward.models.settings import AppSetting

logger = logging.getLogger(__name__)

_DEFAULT_WEBHOOK_TEMPLATE = (
    '{"content": "**{{ alert.state }}** — {{ alert.resource }} — '
    '{{ alert.rule_name }} ({{ alert.metric }} = {{ alert.value }})"}'
)

# All recognised settings and their defaults.
# Plugin settings are stored as "plugin.<name>" and handled separately.
DEFAULTS: dict[str, Any] = {
    # General — external URL for install scripts, share links, alert deep-links.
    # Empty = fall back to the current request's Host header.
    "general.public_base_url": "",
    # Reverse proxies in front of Steward whose X-Forwarded-* headers are
    # believed (steward/core/proxy.py). 0 = trust none.
    "general.trusted_proxy_hops": 1,
    "session.lifetime_hours": 8,
    # Failed logins from one address before it is refused for the window.
    "login.max_failures": 5,
    "login.lockout_minutes": 15,
    # Every growing time-series table (monitor results, Ansible runs and their
    # log files, alert history, Traefik and UniFi samples) keeps this many days.
    "data.retention_days": 90,
    # The audit log is a security record: kept longer than operational data.
    "audit.retention_days": 365,
    "monitors.poll_interval_seconds": 60,
    # Root log level: DEBUG, INFO, WARNING or ERROR (steward/core/logging_setup.py).
    # WARNING also hides the per-request access lines.
    "log.level": "INFO",
    "smtp.host": "",
    "smtp.port": 587,
    "smtp.tls": True,
    "smtp.username": "",
    "smtp.password": "",
    "smtp.recipients": [],
    "smtp.from": "",
    "webhook.url": "",
    "webhook.template": _DEFAULT_WEBHOOK_TEMPLATE,
    "ansible.sources": [],
    # Ansible credentials — global, used by every run (manual + alert-triggered).
    # ssh_private_key/become/vault are encrypted at rest (see SECRET_KEYS).
    "ansible.ssh_private_key": "",
    "ansible.become_password": "",
    "ansible.vault_password": "",
    # Public half of the managed keypair — non-secret, displayed so it can be
    # sprayed onto hosts (provisioning installs it into ~steward/.ssh).
    "ansible.ssh_public_key": "",
    # Default SSH login for steady-state runs — the dedicated account
    # provisioning creates. Acts as a floor (--user); a target's ansible_user
    # or a per-run bootstrap override still wins.
    "ansible.ssh_user": "steward",
    "ansible.host_key_checking": False,
    # Max simultaneous playbook runs; extra runs queue. Applied at app start.
    "ansible.max_concurrent_runs": 3,
    "ping.threshold.good_ms": 50,
    "ping.threshold.warn_ms": 200,
    # Degraded/critical cutoffs for metric coloring (warn=amber, crit=red).
    # cpu/mem/disk/load are percentages (load is load-per-core %); temp in °C.
    # uptime is "higher is better" so its warn/crit are floors. Ping latency
    # reuses ping.threshold.good_ms/warn_ms above.
    "thresholds.cpu_warn": 80, "thresholds.cpu_crit": 90,
    "thresholds.mem_warn": 80, "thresholds.mem_crit": 90,
    "thresholds.disk_warn": 80, "thresholds.disk_crit": 90,
    "thresholds.load_warn": 80, "thresholds.load_crit": 100,
    "thresholds.temp_warn": 70, "thresholds.temp_crit": 85,
    "thresholds.uptime_warn": 99.0, "thresholds.uptime_crit": 95.0,
    # Days left on an HTTPS monitor's certificate (lower is worse), and how
    # many container deaths within an hour count as crash-looping.
    "thresholds.tls_warn_days": 14, "thresholds.tls_crit_days": 3,
    "thresholds.crash_loop_deaths": 3,
    # Seconds a host's clock may be from Steward's before it is called out
    # (milestone 490), either way.
    "thresholds.clock_warn_s": 30, "thresholds.clock_crit_s": 300,
    # Docker time-series retention (rule 25 — tunable, no restart). Raw 30s
    # samples are heavy, so keep a short raw window then roll up to hourly
    # averages kept much longer; lifecycle events are light, keep a month.
    "docker.retention.metrics_raw_days": 7,
    "docker.retention.metrics_rollup_days": 90,
    "docker.retention.events_days": 30,
    # Container logs (m79): on by default for every container (operator
    # preference). `exclude` names containers the server drops on ingest; the
    # per-container ring bounds storage (rotate oldest past whichever of ~age or
    # ~bytes hits first — a chatty container just keeps a shorter window).
    "docker.logs.enabled": True,
    "docker.logs.exclude": [],
    "docker.logs.retention_days": 3,
    "docker.logs.max_bytes_per_container": 5_000_000,
    # Host/plugin metrics retention (plugin_metrics): keep a short raw window at
    # the agent's ~30s cadence, then roll up to hourly averages kept much longer.
    "metrics.retention.raw_days": 7,
    "metrics.retention.rollup_days": 90,
    "plugins.index_url": "https://git.fabledsword.com/bvandeusen/Steward-plugins/raw/branch/main/index.yaml",
    # Default-enabled plugins. These are the generic, non-vendor-specific
    # bundled plugins (protocols/standards, not a single product) — useful on
    # almost any install, so a fresh deployment comes up monitoring rather than
    # blank. Vendor-specific plugins (traefik, unifi) stay opt-in. An operator
    # who disables one writes plugin.<name>={"enabled": False}, which overrides
    # these defaults (stored value wins in get_all_settings/load_settings_sync).
    # Per-plugin yaml config defaults are merged on top at load time.
    "plugin.docker": {"enabled": True},
    "plugin.host_agent": {"enabled": True},
    "plugin.snmp": {"enabled": True},
    # OIDC single-sign-on
    "oidc.enabled": False,
    "oidc.discovery_url": "",
    "oidc.client_id": "",
    "oidc.client_secret": "",
    "oidc.scopes": "openid profile email",
    "oidc.username_claim": "preferred_username",
    "oidc.email_claim": "email",
    "oidc.groups_claim": "groups",
    "oidc.admin_group": "",
    "oidc.operator_group": "",
    # LDAP authentication
    "ldap.enabled": False,
    "ldap.host": "",
    "ldap.port": 389,
    "ldap.tls": False,
    "ldap.ca_cert": "",
    "ldap.bind_dn": "",
    "ldap.bind_password": "",
    "ldap.base_dn": "",
    "ldap.user_filter": "(uid={username})",
    "ldap.admin_group_dn": "",
    "ldap.operator_group_dn": "",
    "ldap.attr_username": "uid",
    "ldap.attr_email": "mail",
    # Scheduled reports
    "reports.enabled": False,
    "reports.schedule_day": 6,   # 0=Monday … 6=Sunday
    "reports.schedule_hour": 8,  # UTC hour
    "reports.last_sent_at": "",
}

# Settings encrypted at rest (transparent encrypt-on-write / decrypt-on-read).
# Adding a key here makes new writes ciphertext; run migrate_plaintext_secrets
# to convert any existing plaintext rows.
SECRET_KEYS: set[str] = {
    "smtp.password",
    "oidc.client_secret",
    "ldap.bind_password",
    "ansible.ssh_private_key",
    "ansible.become_password",
    "ansible.vault_password",
    # A webhook URL carries its own credential (Discord, Slack and ntfy put the
    # token in the path).
    "webhook.url",
}


def _decode(value: Any, key: str = "") -> Any:
    """Decrypt a stored value if it's an encrypted token; else pass through.

    key is passed through to the decrypt log so a wrong-key failure names the
    exact setting that needs re-entering.
    """
    from steward.core.crypto import decrypt_secret, is_encrypted
    return decrypt_secret(value, context=key) if is_encrypted(value) else value


# Secret settings that are stored as ciphertext but won't decrypt with the
# current app key (key rotated or lost). Surfaced as an admin banner so the
# operator knows precisely which secrets to re-enter — instead of finding out
# only when something that uses one fails. Refreshed at startup
# (scan_undecryptable_secrets) and kept live: re-entering a secret clears it
# without a restart (set_setting discards it on a fresh write).
_undecryptable_secrets: set[str] = set()


def get_undecryptable_secrets() -> list[str]:
    """Sorted keys whose stored ciphertext won't decrypt (for the UI banner)."""
    return sorted(_undecryptable_secrets)


def _is_undecryptable(stored: Any, key: str = "") -> bool:
    """True iff `stored` is an encrypted token that fails to decrypt.

    A failed decrypt returns the ciphertext unchanged (still enc-prefixed), so a
    value that is still encrypted after a decrypt attempt is undecryptable.
    """
    from steward.core.crypto import decrypt_secret, is_encrypted
    return (
        isinstance(stored, str)
        and is_encrypted(stored)
        and is_encrypted(decrypt_secret(stored, context=key))
    )


# Git sources live as a list under one setting, so their tokens are sealed
# field by field rather than through SECRET_KEYS.
SOURCES_KEY = "ansible.sources"


def seal_source_tokens(sources: Any) -> tuple[Any, int]:
    """Encrypt each source's plaintext http_token; returns (sources, how many)."""
    from steward.core.crypto import encrypt_secret, is_encrypted
    if not isinstance(sources, list):
        return sources, 0
    sealed, count = [], 0
    for src in sources:
        token = src.get("http_token") if isinstance(src, dict) else None
        if isinstance(token, str) and token and not is_encrypted(token):
            src = {**src, "http_token": encrypt_secret(token)}
            count += 1
        sealed.append(src)
    return sealed, count


def source_token_key(name: str) -> str:
    """How a source's token is named in logs and the undecryptable-secrets banner."""
    return f"{SOURCES_KEY}[{name}].http_token"


# ─────────────────────────────────────────────────────────────────────────────
# Async helpers (use inside request handlers / scheduled tasks)
# ─────────────────────────────────────────────────────────────────────────────

async def get_setting(session: AsyncSession, key: str) -> Any:
    """Return the value for key, or the default if not set."""
    result = await session.execute(
        select(AppSetting).where(AppSetting.key == key)
    )
    row = result.scalar_one_or_none()
    if row is None:
        return DEFAULTS.get(key)
    return _decode(json.loads(row.value_json), key)


async def set_setting(session: AsyncSession, key: str, value: Any) -> None:
    """Upsert a setting (encrypting secret keys at rest). Call in a transaction."""
    to_store = value
    if key in SECRET_KEYS and isinstance(value, str) and value:
        from steward.core.crypto import encrypt_secret
        to_store = encrypt_secret(value)
    result = await session.execute(
        select(AppSetting).where(AppSetting.key == key)
    )
    row = result.scalar_one_or_none()
    now = datetime.now(timezone.utc)
    if row is None:
        session.add(AppSetting(key=key, value_json=json.dumps(to_store), updated_at=now))
    else:
        row.value_json = json.dumps(to_store)
        row.updated_at = now

    # A fresh write of a secret is encrypted with the current key (or cleared to
    # plaintext), so it's decryptable now — drop any stale "undecryptable" flag
    # so the banner clears without a restart.
    if key in SECRET_KEYS:
        _undecryptable_secrets.discard(key)


async def delete_setting(session: AsyncSession, key: str) -> bool:
    """Remove a stored setting row. Returns True if a row was actually deleted.

    Call in a transaction. Deleting a key that also has a DEFAULT reverts it to
    that default rather than unsetting it — so this only meaningfully *removes*
    a setting when no default declares it.
    """
    result = await session.execute(
        select(AppSetting).where(AppSetting.key == key)
    )
    row = result.scalar_one_or_none()
    if row is None:
        return False
    await session.delete(row)
    return True


async def get_stored_plugin_names(session: AsyncSession) -> list[str]:
    """Plugin names that have a real stored row, ignoring DEFAULTS.

    Deliberately NOT derived from get_all_settings: that merges DEFAULTS in, and
    a plugin.* key present only as a default is a code-level declaration rather
    than operator data. Offering to "remove" such a key would be a lie — the
    default would simply reassert it on the next load (exactly the bug behind
    the phantom `plugin.http` entry). Only stored rows can actually be cleaned up.

    Returns names only, never values: plugin config can hold credentials, and
    nothing that lists orphans needs to read them.
    """
    prefix = "plugin."
    result = await session.execute(
        select(AppSetting.key).where(AppSetting.key.like(f"{prefix}%"))
    )
    return sorted(
        key[len(prefix):] for key in result.scalars() if key != prefix
    )


def find_orphaned_plugins(
    stored_names: Iterable[str],
    installed_names: Iterable[str],
    failures: Mapping[str, str] | None = None,
) -> list[dict[str, Any]]:
    """Plugins that are configured or failed to load, but are not installed.

    Pure so it can be tested without a DB or a filesystem. "Not installed" is
    intentionally not treated as "safe to delete" — an external plugin under
    /data/plugins can be missing merely because the volume isn't mounted or an
    install failed, so the caller surfaces these for an explicit operator
    decision instead of removing them automatically.

    Driven by the UNION of stored settings and load failures, not by stored
    settings alone. `load_plugins` records a failure for any enabled plugin
    whose directory it cannot find, and the admin banner counts those — so a
    failure with no stored row (a plugin enabled by a DEFAULTS entry) would
    otherwise be counted in the banner and shown nowhere, which is exactly the
    dead end this section exists to close.

    Each row carries `removable`: only a plugin with a real stored row can
    actually be cleaned up. Offering Remove for a default-declared plugin would
    be a lie — deleting nothing, while the default reasserts it on next load.
    """
    installed = set(installed_names)
    failures = dict(failures or {})
    stored = {name for name in stored_names if name}
    candidates = (stored | set(failures)) - installed
    return [
        {
            "name": name,
            "reason": failures.get(name),
            "removable": name in stored,
        }
        for name in sorted(candidates)
    ]


async def get_all_settings(session: AsyncSession) -> dict[str, Any]:
    """Return flat key→value dict with defaults filled in for missing keys."""
    result = await session.execute(select(AppSetting))
    stored = {row.key: _decode(json.loads(row.value_json), row.key) for row in result.scalars()}
    out: dict[str, Any] = {}
    for key, default in DEFAULTS.items():
        out[key] = stored.get(key, default)
    # Include any plugin.* keys stored in DB
    for key, value in stored.items():
        if key.startswith("plugin.") and key not in out:
            out[key] = value
    return out


# ─────────────────────────────────────────────────────────────────────────────
# Structured config extractors (dict shapes expected by existing consumers)
# ─────────────────────────────────────────────────────────────────────────────

def to_smtp_cfg(settings: dict[str, Any]) -> dict:
    return {
        "host": settings.get("smtp.host", ""),
        "port": settings.get("smtp.port", 587),
        "tls": settings.get("smtp.tls", True),
        "username": settings.get("smtp.username", ""),
        "password": settings.get("smtp.password", ""),
        "recipients": settings.get("smtp.recipients", []),
        "from": settings.get("smtp.from", ""),
    }


def to_webhook_cfg(settings: dict[str, Any]) -> dict:
    return {
        "url": settings.get("webhook.url", ""),
        "template": settings.get("webhook.template", _DEFAULT_WEBHOOK_TEMPLATE),
    }


def to_ansible_cfg(settings: dict[str, Any]) -> dict:
    return {
        "sources": settings.get("ansible.sources", []),
        "ssh_private_key": settings.get("ansible.ssh_private_key", ""),
        "become_password": settings.get("ansible.become_password", ""),
        "vault_password": 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),
    }


def to_oidc_cfg(settings: dict[str, Any]) -> dict:
    return {k[len("oidc."):]: settings.get(k, DEFAULTS[k]) for k in DEFAULTS if k.startswith("oidc.")}


def to_ldap_cfg(settings: dict[str, Any]) -> dict:
    return {k[len("ldap."):]: settings.get(k, DEFAULTS[k]) for k in DEFAULTS if k.startswith("ldap.")}


def _int(settings: dict[str, Any], key: str) -> int:
    try:
        return int(settings.get(key, DEFAULTS[key]))
    except (TypeError, ValueError):
        return int(DEFAULTS[key])


def to_access_cfg(settings: dict[str, Any]) -> dict:
    """app.config keys for sessions, logins and the trusted proxy."""
    from datetime import timedelta
    hours = max(1, _int(settings, "session.lifetime_hours"))
    return {
        "SESSION_LIFETIME_HOURS": hours,
        "PERMANENT_SESSION_LIFETIME": timedelta(hours=hours),
        "TRUSTED_PROXY_HOPS": max(0, _int(settings, "general.trusted_proxy_hops")),
        "LOGIN_MAX_FAILURES": max(1, _int(settings, "login.max_failures")),
        "LOGIN_LOCKOUT_MINUTES": max(1, _int(settings, "login.lockout_minutes")),
    }


def to_thresholds_cfg(settings: dict[str, Any]) -> dict:
    """Per-metric (warn, crit, direction) policy for degraded-value coloring.

    Centralizes "what counts as degraded" so templates just name a metric kind
    via the status_class() and threshold_style() jinja globals. `dir` is
    "high" (higher is worse, e.g. CPU) or "low" (lower is worse, e.g. uptime
    %). Latency reuses the existing ping good/warn thresholds.
    """
    g = settings.get
    return {
        "cpu":     {"warn": g("thresholds.cpu_warn", 80),  "crit": g("thresholds.cpu_crit", 90),  "dir": "high"},
        "mem":     {"warn": g("thresholds.mem_warn", 80),  "crit": g("thresholds.mem_crit", 90),  "dir": "high"},
        "disk":    {"warn": g("thresholds.disk_warn", 80), "crit": g("thresholds.disk_crit", 90), "dir": "high"},
        "load":    {"warn": g("thresholds.load_warn", 80), "crit": g("thresholds.load_crit", 100), "dir": "high"},
        "temp":    {"warn": g("thresholds.temp_warn", 70), "crit": g("thresholds.temp_crit", 85), "dir": "high"},
        "latency": {"warn": g("ping.threshold.good_ms", 50), "crit": g("ping.threshold.warn_ms", 200), "dir": "high"},
        "uptime":  {"warn": g("thresholds.uptime_warn", 99.0), "crit": g("thresholds.uptime_crit", 95.0), "dir": "low"},
        "tls_days": {"warn": g("thresholds.tls_warn_days", 14), "crit": g("thresholds.tls_crit_days", 3), "dir": "low"},
        "clock":   {"warn": g("thresholds.clock_warn_s", 30), "crit": g("thresholds.clock_crit_s", 300), "dir": "high"},
        "crash_loop": {"warn": g("thresholds.crash_loop_deaths", 3), "crit": g("thresholds.crash_loop_deaths", 3), "dir": "high"},
    }


_STATUS_STYLE = {"warn": "color:var(--yellow);font-weight:500;",
                 "crit": "color:var(--red);font-weight:500;"}


def threshold_style_for(value: Any, kind: str, thresholds: dict) -> str:
    """Return an inline-style fragment (amber at warn, red at crit) for a metric,
    graded by formatting.status_class against `thresholds` (to_thresholds_cfg).
    Empty string when the reading is normal or missing — drop straight into a
    span's style="".
    """
    from steward.core.formatting import status_class
    return _STATUS_STYLE.get(status_class(value, kind, thresholds), "")


def to_plugins_cfg(settings: dict[str, Any]) -> dict:
    """Assemble {plugin_name: {...config}} from all plugin.* keys."""
    result = {}
    for key, value in settings.items():
        if key.startswith("plugin."):
            name = key[len("plugin."):]
            result[name] = value
    return result


# ─────────────────────────────────────────────────────────────────────────────
# Synchronous loader — safe to call before the event loop starts
# ─────────────────────────────────────────────────────────────────────────────

def load_settings_sync(db_url: str) -> dict[str, Any]:
    """Load all settings from DB synchronously via asyncio.run().

    Safe to call in create_app() before the Quart event loop starts.
    Returns flat key→value dict with defaults filled in.
    """
    async def _load() -> dict[str, Any]:
        from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
        engine = create_async_engine(db_url, echo=False)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        try:
            async with factory() as session:
                result = await session.execute(select(AppSetting))
                return {row.key: _decode(json.loads(row.value_json), row.key) for row in result.scalars()}
        finally:
            await engine.dispose()

    stored = asyncio.run(_load())
    out: dict[str, Any] = {}
    for key, default in DEFAULTS.items():
        out[key] = stored.get(key, default)
    for key, value in stored.items():
        if key.startswith("plugin.") and key not in out:
            out[key] = value
    return out


def migrate_plaintext_secrets(db_url: str) -> int:
    """Encrypt any existing plaintext secret rows in place. Idempotent.

    Returns the number of values converted. Run once at startup after the
    encryptor is initialised (already-encrypted rows are skipped).
    """
    from steward.core.crypto import encrypt_secret, is_encrypted

    async def _run() -> int:
        from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
        engine = create_async_engine(db_url, echo=False)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        converted = 0
        try:
            async with factory() as session:
                async with session.begin():
                    result = await session.execute(
                        select(AppSetting).where(AppSetting.key.in_(SECRET_KEYS))
                    )
                    for row in result.scalars():
                        val = json.loads(row.value_json)
                        if isinstance(val, str) and val and not is_encrypted(val):
                            row.value_json = json.dumps(encrypt_secret(val))
                            converted += 1
                    row = (await session.execute(
                        select(AppSetting).where(AppSetting.key == SOURCES_KEY))).scalar_one_or_none()
                    if row is not None:
                        sources, sealed = seal_source_tokens(json.loads(row.value_json))
                        if sealed:
                            row.value_json = json.dumps(sources)
                            converted += sealed
        finally:
            await engine.dispose()
        return converted

    return asyncio.run(_run())


def scan_undecryptable_secrets(db_url: str) -> list[str]:
    """Populate the undecryptable-secrets cache from the DB; return the keys found.

    Run once at startup, AFTER migrate_plaintext_secrets and init_crypto: any row
    that's encrypted but won't decrypt with the current key was sealed under a
    different (lost/rotated) key and must be re-entered. Surfaced via the admin
    banner (get_undecryptable_secrets).
    """
    async def _run() -> list[str]:
        from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
        engine = create_async_engine(db_url, echo=False)
        factory = async_sessionmaker(engine, expire_on_commit=False)
        bad: list[str] = []
        try:
            async with factory() as session:
                result = await session.execute(
                    select(AppSetting).where(AppSetting.key.in_(SECRET_KEYS))
                )
                for row in result.scalars():
                    if _is_undecryptable(json.loads(row.value_json), row.key):
                        bad.append(row.key)
                row = (await session.execute(
                    select(AppSetting).where(AppSetting.key == SOURCES_KEY))).scalar_one_or_none()
                sources = json.loads(row.value_json) if row is not None else []
                for src in sources if isinstance(sources, list) else []:
                    if isinstance(src, dict):
                        key = source_token_key(src.get("name", "?"))
                        if _is_undecryptable(src.get("http_token"), key):
                            bad.append(key)
        finally:
            await engine.dispose()
        return bad

    global _undecryptable_secrets
    found = asyncio.run(_run())
    _undecryptable_secrets = set(found)
    return sorted(found)


# ─────────────────────────────────────────────────────────────────────────────
# External URL helper
# ─────────────────────────────────────────────────────────────────────────────

def public_base_url(request) -> str:
    """Return the externally-reachable base URL for this Steward instance.

    Prefers the admin-configured 'general.public_base_url' setting (cached in
    current_app.config['PUBLIC_BASE_URL']) when set. Falls back to
    request.host_url stripped of its trailing slash, so unconfigured
    single-hostname installs keep working with zero config.

    Use this — NOT request.host_url — anywhere you build a URL that will be
    consumed by something outside this Quart request: install scripts, share
    links, alert notifications, webhook callbacks. The Host header is not
    reliable behind proxies or on multi-hostname deployments.
    """
    from quart import current_app
    configured = (current_app.config.get("PUBLIC_BASE_URL") or "").strip()
    if configured:
        return configured.rstrip("/")
    return request.host_url.rstrip("/")