haproxy-openmanager/backend/services/acme_diagnostics.py

"""
ACME Diagnostics service (Feature A — Issue #13).

Pre-flight & post-failure diagnostics for an ACME order. Each check produces a
structured `{id, label, status, message, details, duration_ms, severity}` row
suitable for an Antd Tabs/Steps display.

Key constraints (Section 3.3 of the v1.5.0 plan):
- DNS resolution uses stdlib socket.gethostbyname_ex via run_in_executor (we
  intentionally avoid pulling aiodns as a runtime dep for v1.5.0).
- Port-80 probe is a GET (not HEAD) locked to the order's domains, because the
  status code alone cannot tell a working challenge endpoint from a web UI: a
  reverse proxy that has lost its /.well-known/acme-challenge/ location serves
  its SPA with HTTP 200. The body's shape decides. Warns rather than fails on
  egress timeout (corp egress policies often blackhole outbound 80) and on a
  wrong responder (this probe sees the PUBLIC domain, not the challenge backend,
  so it is evidence rather than a verdict).
- All checks have hard wall-clock timeouts (asyncio.wait_for) to bound impact
  on the API event loop.
- humanize_error_detail covers >= 11 RFC8555 problem types and is backwards
  compatible with the legacy plain-string error_detail field.
"""

import asyncio
import ipaddress
import json
import logging
import socket
import time
from typing import Any, Dict, List, Optional

import aiohttp

logger = logging.getLogger(__name__)


# R18b audit fix (round 4 #B): SSRF guard for outbound HTTP probes.
# The ACME diagnostics check_port80 helper opens an HTTP HEAD against
# the operator-supplied domain. If that domain resolves to a private
# / loopback / link-local / cloud-metadata IP, the API host becomes a
# request-forwarding primitive: an authenticated operator could
# fingerprint internal services or hit AWS/GCP metadata endpoints by
# pointing DNS at them. Refuse to probe non-public IPs and surface
# the skip in the diagnostic result so the operator knows why.
def _is_public_ip(ip_str: str) -> bool:
    """Return True only for globally-routable IPv4/IPv6 addresses.

    Excludes loopback, link-local, RFC1918 private space, multicast,
    cloud-metadata IPs (169.254.169.254 falls under link-local), and
    reserved blocks. Used by check_port80 before issuing an HTTP
    request to operator-supplied hostnames.
    """
    try:
        ip = ipaddress.ip_address(ip_str)
    except (ValueError, TypeError):
        return False
    # R18c audit fix (round 3 #4 — KRITIK SSRF): normalize IPv4-mapped
    # IPv6 addresses to their underlying IPv4 form before
    # classification. PRE-FIX an attacker who controlled the
    # domain's AAAA record could point it at `::ffff:127.0.0.1`
    # (or `::ffff:169.254.169.254` for cloud metadata) and our
    # guard would return True because IPv6Address.is_loopback /
    # is_private only check the IPv6 address space — they do NOT
    # walk into the embedded IPv4 mapping. The guard would then
    # let the probe through, creating an SSRF path back into the
    # OpenManager host's loopback / cloud metadata service. Always
    # unwrap `.ipv4_mapped` first so the IPv4 classification rules
    # apply.
    if isinstance(ip, ipaddress.IPv6Address) and ip.ipv4_mapped is not None:
        ip = ip.ipv4_mapped
    if ip.is_loopback or ip.is_link_local or ip.is_private:
        return False
    if ip.is_multicast or ip.is_reserved or ip.is_unspecified:
        return False
    return True


async def _all_ips_public(domain: str, *, timeout: float = 5.0) -> tuple[bool, list[str]]:
    """Resolve `domain` and return (all_public, ips). On DNS failure
    returns (False, []) — caller should treat as "skip / unable to
    verify safety" rather than "probe anyway"."""
    try:
        info = await _resolve_dns(domain, timeout=timeout)
    except Exception:
        return (False, [])
    ips = info.get("ips", []) or []
    if not ips:
        return (False, [])
    return (all(_is_public_ip(ip) for ip in ips), ips)


# RFC8555 problem types (https://datatracker.ietf.org/doc/html/rfc8555#section-6.7)
# Plus a few extra ACMEv2 additions used in the wild.
_PROBLEM_HUMANIZED: Dict[str, Dict[str, str]] = {
    "urn:ietf:params:acme:error:accountDoesNotExist": {
        "title": "ACME account not found",
        "hint": "The ACME account is missing or has been deactivated. Re-create the LE account from Settings → Let's Encrypt.",
    },
    "urn:ietf:params:acme:error:badNonce": {
        "title": "Stale request nonce",
        "hint": "Transient — the next retry should succeed. If it persists, your system clock may be skewed.",
    },
    "urn:ietf:params:acme:error:badRevocationReason": {
        "title": "Invalid revocation reason",
        "hint": "The CA rejected the revocation reason code. Use a valid RFC5280 CRLReason.",
    },
    "urn:ietf:params:acme:error:caa": {
        "title": "CAA record forbids issuance",
        "hint": "DNS CAA records prevent Let's Encrypt from issuing this certificate. Add 'letsencrypt.org' to the CAA records.",
    },
    "urn:ietf:params:acme:error:connection": {
        "title": "CA could not connect to your server",
        "hint": "Let's Encrypt's validators could not reach port 80 from the public internet. Check inbound firewall and routing.",
    },
    "urn:ietf:params:acme:error:dns": {
        "title": "DNS resolution failed during validation",
        "hint": "The domain does not resolve, or the CA's DNS lookup timed out. Verify A/AAAA records are public.",
    },
    "urn:ietf:params:acme:error:incorrectResponse": {
        "title": "HTTP-01 challenge response mismatch",
        "hint": "The CA fetched the challenge URL but received the wrong key authorization. Confirm the challenge was served from the right backend.",
    },
    "urn:ietf:params:acme:error:externalAccountRequired": {
        "title": "External Account Binding (EAB) required",
        "hint": "This CA (e.g. ZeroSSL, Google) requires EAB. Enter the EAB Key ID and HMAC Key from your CA account when registering.",
    },
    "urn:ietf:params:acme:error:invalidContact": {
        "title": "Invalid contact email",
        "hint": "The ACME account email is malformed. Update the LE account email.",
    },
    "urn:ietf:params:acme:error:malformed": {
        "title": "Malformed request",
        "hint": "The request body could not be parsed. Often a transient bug — retry; if it persists, raise an issue.",
    },
    "urn:ietf:params:acme:error:rateLimited": {
        "title": "Let's Encrypt rate limit hit",
        "hint": "Too many certificates issued or too many duplicate orders. Wait or use the staging directory.",
    },
    "urn:ietf:params:acme:error:rejectedIdentifier": {
        "title": "Domain rejected by CA",
        "hint": "The CA refused this hostname (e.g. blocklisted TLD, public-suffix mismatch).",
    },
    "urn:ietf:params:acme:error:serverInternal": {
        "title": "ACME server error",
        "hint": "Let's Encrypt is reporting a transient server error. Retry.",
    },
    "urn:ietf:params:acme:error:tls": {
        "title": "TLS error during validation",
        "hint": "The validator could not complete the TLS handshake (only relevant for tls-alpn-01 / tls-sni).",
    },
    "urn:ietf:params:acme:error:unauthorized": {
        "title": "Unauthorized",
        "hint": "The challenge response could not be verified — most often an HTTP-01 path-not-served issue.",
    },
    "urn:ietf:params:acme:error:unsupportedContact": {
        "title": "Unsupported contact scheme",
        "hint": "Only 'mailto:' contacts are currently supported by Let's Encrypt.",
    },
    "urn:ietf:params:acme:error:unsupportedIdentifier": {
        "title": "Unsupported identifier",
        "hint": "Only DNS identifiers are supported.",
    },
    "urn:ietf:params:acme:error:userActionRequired": {
        "title": "User action required",
        "hint": "ACME account requires Terms-of-Service re-acceptance. Visit the URL in the error to acknowledge.",
    },
}


def humanize_error_detail(error_detail: Any) -> Dict[str, Any]:
    """Convert the order.error_detail field into a UI-friendly structured form.

    error_detail may be:
      - A JSON string with {type, detail, status, subproblems}
      - A plain string (legacy)
      - None
    Always returns a dict with at minimum {title, message, hint}.
    """
    if not error_detail:
        return {"title": "No error", "message": "", "hint": ""}

    parsed: Optional[Dict[str, Any]] = None
    if isinstance(error_detail, dict):
        parsed = error_detail
    elif isinstance(error_detail, str):
        s = error_detail.strip()
        if s.startswith("{"):
            try:
                parsed = json.loads(s)
            except json.JSONDecodeError:
                parsed = None

    if parsed is None:
        # Legacy plain string fallback
        return {
            "title": "ACME error",
            "message": str(error_detail),
            "hint": "",
            "raw": str(error_detail),
        }

    problem_type = parsed.get("type") or ""
    base = _PROBLEM_HUMANIZED.get(problem_type, {})
    title = base.get("title") or "ACME error"
    hint = base.get("hint") or ""
    message = parsed.get("detail") or parsed.get("message") or ""
    status = parsed.get("status")
    subproblems = parsed.get("subproblems") or []

    # Issue #35: DNS-01 failures are recorded as {stage, reason, timestamp} (no RFC8555 "type"),
    # so without this fallback the humanized alert would show a bare "ACME error" with no message.
    # Surface the reason and a targeted hint so the operator knows exactly what to fix.
    if not problem_type and parsed.get("reason"):
        reason = str(parsed.get("reason"))
        message = message or reason
        title = "DNS-01 validation failed"
        rlow = reason.lower()
        if "decrypt" in rlow or "credential" in rlow:
            hint = hint or "Re-enter the DNS provider credentials for this account in ACME Automation."
        elif "zone" in rlow:
            hint = hint or "Confirm the domain's DNS zone is managed by the configured provider and the token has access to it."
        elif "deadline" in rlow or "expired" in rlow or "confirm" in rlow:
            hint = hint or "The manual confirmation window passed. Create a new certificate request and publish the TXT record promptly."
        else:
            hint = hint or "Check the DNS TXT record and provider credentials, then retry."

    out = {
        "title": title,
        "message": message,
        "hint": hint,
        "type": problem_type,
        "raw": parsed,
    }
    if status is not None:
        out["status"] = status
    if subproblems:
        out["subproblems"] = [
            {
                "type": sp.get("type"),
                "detail": sp.get("detail"),
                "identifier": (sp.get("identifier") or {}).get("value"),
            }
            for sp in subproblems
            if isinstance(sp, dict)
        ]
    return out


# ----------------------------------------------------------------------------
# Per-check helpers
# ----------------------------------------------------------------------------


def _check_result(
    check_id: str,
    label: str,
    status: str,
    message: str,
    *,
    severity: str = "info",
    details: Optional[Dict[str, Any]] = None,
    duration_ms: Optional[int] = None,
) -> Dict[str, Any]:
    return {
        "id": check_id,
        "label": label,
        "status": status,  # 'ok' | 'warn' | 'fail' | 'skipped'
        "severity": severity,  # 'info' | 'warn' | 'error'
        "message": message,
        "details": details or {},
        "duration_ms": duration_ms,
    }


async def _resolve_dns(domain: str, *, timeout: float = 5.0) -> Dict[str, Any]:
    """Resolve a domain via stdlib socket.gethostbyname_ex; never blocks the
    asyncio event loop.
    """
    loop = asyncio.get_running_loop()
    try:
        result = await asyncio.wait_for(
            loop.run_in_executor(None, socket.gethostbyname_ex, domain),
            timeout=timeout,
        )
        canonical, aliases, ips = result
        return {"canonical": canonical, "aliases": aliases, "ips": ips}
    except asyncio.TimeoutError:
        raise
    except Exception as e:
        # socket.gaierror, etc.
        raise RuntimeError(str(e)) from e


async def check_dns(domains: List[str]) -> Dict[str, Any]:
    """Check that each order domain resolves to at least one public-looking IPv4."""
    started = time.time()
    failed: List[Dict[str, Any]] = []
    resolved: Dict[str, List[str]] = {}
    for d in domains:
        # Wildcards are valid per RFC8555 but cannot be HTTP-01 validated; skip
        # actual DNS resolution for them (they would fail A-record lookup).
        if d.startswith("*."):
            resolved[d] = []
            continue
        try:
            r = await _resolve_dns(d, timeout=5.0)
            resolved[d] = r["ips"]
            if not r["ips"]:
                failed.append({"domain": d, "reason": "no A records"})
        except asyncio.TimeoutError:
            failed.append({"domain": d, "reason": "dns timeout"})
        except Exception as e:
            failed.append({"domain": d, "reason": str(e)})

    duration_ms = int((time.time() - started) * 1000)
    if failed:
        return _check_result(
            "dns",
            "DNS resolution",
            "fail",
            f"DNS lookup failed for {len(failed)} domain(s)",
            severity="error",
            details={"failed": failed, "resolved": resolved},
            duration_ms=duration_ms,
        )
    return _check_result(
        "dns",
        "DNS resolution",
        "ok",
        f"All {len(domains)} domain(s) resolved",
        severity="info",
        details={"resolved": resolved},
        duration_ms=duration_ms,
    )


# Enough to classify a response without turning the diagnostic into a way to pull
# arbitrary amounts of a third party's content into our JSON.
_PROBE_BODY_LIMIT = 65536


def _classify_probe_body(body: bytes, content_type: str) -> str:
    """Coarse shape of a probe response: html | json | text | empty | binary.

    The body itself is deliberately NOT retained anywhere — the shape is all that is
    needed to tell "served me a web page" from "served me a token", and keeping the
    bytes would open a new read surface onto whatever is behind the address.
    """
    if not body:
        return "empty"
    ct = (content_type or "").lower()
    head = body[:512].lstrip().lower()
    if ct.startswith("text/html") or head.startswith((b"<!doctype", b"<html")):
        return "html"
    if ct.startswith("application/json") or head[:1] in (b"{", b"["):
        return "json"
    try:
        body.decode("utf-8")
    except UnicodeDecodeError:
        return "binary"
    return "text"


async def check_port80(domains: List[str], *, http_timeout: float = 5.0) -> Dict[str, Any]:
    """Probe HTTP-01 readiness on port 80 with a GET to a synthetic challenge URL.

    404 means the path is served but no challenge is outstanding, which is fine. A
    200 is only fine if the body is NOT a web page: a proxy that has lost its
    /.well-known/acme-challenge/ route falls through to its catch-all and answers
    200 with index.html, which a status-code-only check accepts as healthy while
    every real validation fails.

    On egress timeout we WARN rather than FAIL because many corporate egress
    policies blackhole port 80 outbound; that does not impair LE's ingress
    validation (LE comes inbound).
    """
    started = time.time()
    targets: List[Dict[str, Any]] = []
    timeout = aiohttp.ClientTimeout(total=http_timeout)
    skip_reason = None

    domains_to_check = [d for d in domains if not d.startswith("*.")]
    if not domains_to_check:
        return _check_result(
            "port80",
            "Port 80 reachability",
            "skipped",
            "All domains are wildcards; HTTP-01 not applicable",
            severity="info",
            duration_ms=int((time.time() - started) * 1000),
        )

    # R18c audit fix (round 4 #4 — KRITIK SSRF residual): force the
    # aiohttp connector to family=AF_INET (IPv4-only) so the HTTP
    # probe resolves and connects with the SAME family that
    # _resolve_dns / _all_ips_public classifies. PRE-FIX the SSRF
    # guard ran on the IPv4 list returned by `gethostbyname_ex`,
    # but aiohttp's default connector did its own dual-stack
    # `getaddrinfo` and could connect via AAAA — so an attacker
    # who controlled a domain's DNS could publish a benign public
    # A record (passing our guard) AND a `::1`/`fc00::/7`/`fe80::/10`
    # AAAA record that aiohttp picked, hitting our internal IPv6
    # space. Constraining the connector to IPv4 closes the loop
    # because the family the guard inspects equals the family the
    # connector uses. ACME HTTP-01 itself works only over IPv4-or-
    # IPv6 paths the CA can reach; the diagnostic just needs to
    # confirm reachability and we already only classify IPv4.
    connector = aiohttp.TCPConnector(family=socket.AF_INET, ssl=False)
    async with aiohttp.ClientSession(timeout=timeout, connector=connector) as session:
        for d in domains_to_check:
            # R18b audit fix (round 4 #B — SSRF guard): refuse to
            # probe a domain whose A/AAAA records point at private,
            # loopback, link-local, multicast, or cloud-metadata IP
            # space. Pre-fix the diagnostic was a usable SSRF
            # primitive for any authenticated operator: pick a
            # hostname pointing at 169.254.169.254 / 10.0.0.0/8 /
            # 127.0.0.1 and the API host issued an outbound HEAD,
            # reflecting status / error back into the diagnostic
            # JSON. The HTTP-01 protocol fundamentally requires the
            # CA to reach the host from the public internet, so a
            # private-IP domain cannot validate anyway.
            all_public, ips = await _all_ips_public(d, timeout=2.0)
            if not all_public:
                targets.append({
                    "domain": d,
                    "skip": "non-public IP — refusing to probe (SSRF guard)",
                    "ips": ips,
                    "ok": False,
                    "warn": True,
                })
                if ips:
                    skip_reason = "non-public IPs blocked"
                continue
            url = f"http://{d}/.well-known/acme-challenge/diagnostic-probe"
            try:
                # v1.11.0: recorded as an outbound row so a failing port-80 probe is
                # diagnosable after the fact, not only while the panel is open.
                #
                # MERGE NOTE: the feature branch instrumented a `session.head(...)`
                # probe, which is what this function did when that branch was cut.
                # It has since become a GET, because the status code alone cannot
                # tell a working challenge endpoint from a SPA catch-all serving
                # index.html with HTTP 200 (v1.10.x). Reverting to HEAD to gain the
                # log row would put that bug straight back, so the GET probe below
                # is the one that is wrapped.
                #
                # GET, not HEAD: the status code alone cannot tell a working challenge
                # endpoint from a SPA. A reverse proxy that has lost its
                # /.well-known/acme-challenge/ location falls through to its catch-all
                # and serves index.html with HTTP 200 — which the old
                # `status in (200, 404)` rule accepted as healthy while every real
                # validation failed. Only the body distinguishes them.
                from utils.http_instrumentation import outbound_span, TARGET_ACME_DIAG

                async with outbound_span(
                    target=TARGET_ACME_DIAG, method="GET", url=url, capture_body=False
                ) as span:
                    async with session.get(url, allow_redirects=False) as resp:
                        # The body is EVIDENCE, not a precondition. If it cannot be read —
                        # connection reset mid-response, a server that hangs after headers —
                        # fall back to the status-only semantics this check has always had
                        # rather than turning a healthy 404 into a hard failure. The stricter
                        # rule below applies only when there is something to judge.
                        try:
                            body = await resp.content.read(_PROBE_BODY_LIMIT)
                        except Exception:
                            body = None
                        content_type = (resp.headers.get("content-type") or "").split(";")[0].strip()
                        body_class = (
                            _classify_probe_body(body, content_type) if body is not None else "unread"
                        )
                        target = {
                            "domain": d,
                            "status": resp.status,
                            "content_type": content_type or None,
                            "body_len": len(body) if body is not None else None,
                            "body_class": body_class,
                        }
                        # The VERDICT goes in the log row, not the body. `body` here is
                        # up to 64 KB of a third party's page (`_PROBE_BODY_LIMIT`);
                        # storing it would put an arbitrary remote document into the
                        # audit table for every probed domain. The classification is
                        # what an operator reads back anyway.
                        span.set_response(
                            resp.status,
                            getattr(resp, "headers", None),
                            {
                                "probe": "acme-http01",
                                "content_type": content_type or None,
                                "body_len": len(body) if body is not None else None,
                                "body_class": body_class,
                            },
                        )
                        if resp.status == 200 and body_class == "html":
                            # Reachable, wrong responder. Reported as a warning rather than
                            # a failure: this check probes the PUBLIC domain and cannot see
                            # the challenge backend, so it is evidence, not a verdict — and
                            # a new `fail` here would block the site wizard on upgrade day
                            # for every install.
                            target["warn"] = True
                            target["diagnosis"] = (
                                "responded 200 with an HTML page, not a challenge token — "
                                "the request is reaching a web UI instead of the ACME endpoint"
                            )
                        elif resp.status in (301, 302, 303, 307, 308):
                            target["warn"] = True
                            target["redirect_location"] = resp.headers.get("location")
                            target["diagnosis"] = (
                                "redirected instead of serving the challenge path"
                            )
                        else:
                            target["ok"] = resp.status in (200, 404)
                        targets.append(target)
            except asyncio.TimeoutError:
                targets.append({"domain": d, "error": "egress timeout", "warn": True})
                skip_reason = "egress timeout"
            except aiohttp.ClientError as e:
                targets.append({"domain": d, "error": str(e), "ok": False})
            except Exception as e:
                targets.append({"domain": d, "error": str(e), "ok": False})

    duration_ms = int((time.time() - started) * 1000)
    failed = [t for t in targets if not t.get("ok") and not t.get("warn")]
    warns = [t for t in targets if t.get("warn")]
    if failed:
        return _check_result(
            "port80",
            "Port 80 reachability",
            "fail",
            f"Port 80 probe failed for {len(failed)} domain(s)",
            severity="error",
            details={"targets": targets},
            duration_ms=duration_ms,
        )
    # Surface warnings even when OTHER domains answered correctly. The old condition
    # ("warn only if nothing succeeded") hid the single most diagnostic outcome there
    # is: a multi-domain certificate where one name reaches a web UI instead of the
    # challenge endpoint reported a clean pass.
    if warns:
        # R18b audit fix (round 7): branch the rollup message on the
        # actual cause. Pre-fix the message was always "Egress to
        # port 80 appears blocked" — even when every target was
        # skipped because the SSRF guard refused to probe a non-
        # public IP, which has nothing to do with egress firewalls.
        # Operators saw "egress blocked" and started spelunking
        # corporate firewall logs while the real cause was an
        # internal-only DNS A record. Also harden against
        # `skip_reason=None` so the message never reads "(None)".
        # Wrong-responder warnings take priority over every other cause: they are the
        # only ones that mean "your server answered, and answered wrong", which is a
        # different problem from "we could not test".
        wrong_responder = [t for t in targets if t.get("diagnosis")]
        if wrong_responder:
            first = wrong_responder[0]
            if first.get("body_class") == "html":
                human = (
                    f"{first['domain']} answered HTTP {first.get('status')} with an HTML "
                    f"page ({first.get('content_type') or 'unknown type'}, "
                    f"{first.get('body_len')} bytes) instead of a challenge token. The "
                    "path is reaching a web interface, not the ACME endpoint — check "
                    "that the reverse proxy in front of OpenManager routes "
                    "/.well-known/acme-challenge/ to the API."
                )
            else:
                human = (
                    f"{first['domain']} answered HTTP {first.get('status')} "
                    f"({first.get('diagnosis')})"
                )
            return _check_result(
                "port80",
                "Port 80 reachability",
                "warn",
                human,
                severity="warn",
                details={"targets": targets},
                duration_ms=duration_ms,
            )

        ssrf_skip = any(
            "non-public" in (t.get("skip") or "")
            or "SSRF" in (t.get("skip") or "")
            for t in targets
        )
        if ssrf_skip and not skip_reason:
            skip_reason = "non-public IPs blocked"
        if ssrf_skip:
            human = (
                f"Probe skipped for non-public IPs ({skip_reason}). "
                "ACME HTTP-01 requires a public A record; corporate / "
                "internal-only domains cannot satisfy LE validation."
            )
        else:
            reason = skip_reason or "egress restriction"
            human = (
                f"Egress to port 80 appears blocked ({reason}); "
                "inbound CA validation may still succeed"
            )
        return _check_result(
            "port80",
            "Port 80 reachability",
            "warn",
            human,
            severity="warn",
            details={"targets": targets},
            duration_ms=duration_ms,
        )
    return _check_result(
        "port80",
        "Port 80 reachability",
        "ok",
        f"All {len(domains_to_check)} domain(s) responded on port 80",
        severity="info",
        details={"targets": targets},
        duration_ms=duration_ms,
    )


async def check_routing(conn, domains: List[str], cluster_ids: List[int]) -> Dict[str, Any]:
    """Verify that at least one frontend on the order's clusters has a
    use_backend_rules / acl_rules pointing at the system ACME challenge
    backend OR that an HTTP frontend covering port 80 exists for the
    requesting cluster(s).
    """
    started = time.time()
    if not cluster_ids:
        return _check_result(
            "routing",
            "HAProxy routing",
            "warn",
            "Order has no associated cluster",
            severity="warn",
            duration_ms=int((time.time() - started) * 1000),
        )

    # The WHERE clause is deliberately identical to the pre-existing one, so `not rows`
    # still means exactly what it meant before and the `fail` branch below cannot fire
    # in any situation where it previously passed. Narrowing it here (e.g. by adding a
    # mode filter) would turn a tcp-only port-80 cluster from "ok" into "fail", and the
    # site wizard blocks submit on any failing check — locking those installs the day
    # this ships. Mode is examined afterwards, in Python, and only ever downgrades to
    # `warn`.
    rows = await conn.fetch(
        """
        SELECT f.id, f.name, f.bind_address, f.bind_port, f.mode, f.default_backend,
               f.cluster_id, c.acme_enabled
        FROM frontends f
        JOIN haproxy_clusters c ON c.id = f.cluster_id
        WHERE f.cluster_id = ANY($1::int[]) AND f.is_active = TRUE AND f.bind_port = 80
        """,
        cluster_ids,
    )
    duration_ms = int((time.time() - started) * 1000)
    if not rows:
        return _check_result(
            "routing",
            "HAProxy routing",
            "fail",
            "No HTTP frontend on port 80 found in target cluster(s)",
            severity="error",
            details={"cluster_ids": cluster_ids},
            duration_ms=duration_ms,
        )

    # Character-for-character the renderer's normalisation (services/haproxy_config.py),
    # so this can never disagree with what actually gets emitted.
    http_rows = [r for r in rows if (r["mode"] or "http").strip().lower() == "http"]
    if not http_rows:
        return _check_result(
            "routing",
            "HAProxy routing",
            "warn",
            (
                "The only port-80 frontend(s) in the target cluster(s) are in tcp mode. "
                "A tcp-mode frontend cannot carry the /.well-known/acme-challenge/ ACL, "
                "so HTTP-01 cannot be served — use DNS-01, or add an http-mode frontend "
                "on port 80."
            ),
            severity="warn",
            details={"frontends": [dict(r) for r in rows]},
            duration_ms=duration_ms,
        )
    rows = http_rows

    # A frontend row proves only that the DATABASE describes port-80 routing. The
    # renderer gates the challenge ACL on `acme_enabled`, and the nodes run whatever
    # config was last APPLIED — so the row said "ok" during an incident where the
    # live config had no usable challenge route at all. Check the two things the row
    # cannot tell us. Both report `warn`, never `fail`: the site wizard blocks submit
    # on any `fail`, so a new failing condition would lock every install on the day
    # it ships.
    # `.get()` rather than `[]`: the column gates a WARNING, so a row shape without
    # it should not blow up the whole diagnostic. Absent means "assume enabled" —
    # the applied-config check below is the authoritative one either way.
    acme_off = sorted({r["cluster_id"] for r in rows if not r.get("acme_enabled", True)})
    if acme_off:
        return _check_result(
            "routing",
            "HAProxy routing",
            "warn",
            (
                f"Cluster(s) {acme_off} have ACME Challenge Routing disabled, so the "
                "generated config contains no /.well-known/acme-challenge/ route. "
                "Enable it in Cluster Management, then apply the cluster."
            ),
            severity="warn",
            details={"frontends": [dict(r) for r in rows], "acme_disabled_clusters": acme_off},
            duration_ms=duration_ms,
        )

    missing_in_applied = []
    challenge_backends = {}
    for cluster_id in sorted({r["cluster_id"] for r in rows}):
        # Selector matched to the one the AGENT uses to fetch its config
        # (routers/agent.py: status='APPLIED' AND is_active=TRUE), because the question
        # here is "what are the nodes running right now?". Without `is_active` this can
        # read a superseded row and report on a config that was never delivered.
        # Extracting in SQL rather than pulling whole configs back per cluster: these
        # files run to hundreds of KB on real installs.
        applied = await conn.fetchrow(
            """
            SELECT position('use_backend _acme_challenge_backend' in config_content) > 0
                       AS has_route,
                   substring(config_content from 'server _acme_mgmt [^\\n]*') AS server_line
            FROM config_versions
            WHERE cluster_id = $1 AND status = 'APPLIED' AND is_active = TRUE
              AND config_content IS NOT NULL
            ORDER BY created_at DESC LIMIT 1
            """,
            cluster_id,
        )
        if not applied or not applied["has_route"]:
            missing_in_applied.append(cluster_id)
            continue
        server_line = (applied["server_line"] or "").strip()
        challenge_backends[cluster_id] = (
            server_line[len("server _acme_mgmt "):].strip() if server_line else None
        )

    if missing_in_applied:
        return _check_result(
            "routing",
            "HAProxy routing",
            "warn",
            (
                f"Cluster(s) {missing_in_applied} have no applied configuration carrying "
                "the challenge route. The change exists in the database but the nodes are "
                "still running an older config — apply the cluster."
            ),
            severity="warn",
            details={"frontends": [dict(r) for r in rows], "clusters_not_applied": missing_in_applied},
            duration_ms=duration_ms,
        )

    # A backend section with no `server` line: the route exists, `haproxy -c` passes,
    # and every challenge request gets a 503 from an empty backend. Without this branch
    # the falsy target slips past the loopback filter below and the check reports "ok".
    serverless = sorted(cid for cid, target in challenge_backends.items() if not target)
    if serverless:
        return _check_result(
            "routing",
            "HAProxy routing",
            "warn",
            (
                f"Cluster(s) {serverless} route the challenge path to a backend that has "
                "no server line, so every request returns 503. The configured ACME "
                "Challenge Backend URL could not be resolved into an address — check it "
                "in Cluster Management, or Settings > ACME for the global value."
            ),
            severity="warn",
            details={"frontends": [dict(r) for r in rows], "clusters_without_server": serverless},
            duration_ms=duration_ms,
        )

    loopback = {
        cid: target for cid, target in challenge_backends.items()
        if target and target.split(":")[0].strip("[]").lower()
        in ("localhost", "127.0.0.1", "::1", "0.0.0.0")
    }
    if loopback:
        return _check_result(
            "routing",
            "HAProxy routing",
            "warn",
            (
                f"The applied config points the challenge backend at {sorted(loopback.values())}. "
                "HAProxy resolves that on the HAProxy node, so it means the node itself, not "
                "this management server. Set ACME Challenge Backend URL to a routable address."
            ),
            severity="warn",
            details={"frontends": [dict(r) for r in rows], "challenge_backends": challenge_backends},
            duration_ms=duration_ms,
        )

    return _check_result(
        "routing",
        "HAProxy routing",
        "ok",
        f"Found {len(rows)} HTTP frontend(s) on port 80; challenge route present in applied config",
        severity="info",
        details={"frontends": [dict(r) for r in rows], "challenge_backends": challenge_backends},
        duration_ms=duration_ms,
    )


async def check_account(conn, account_id: Optional[int]) -> Dict[str, Any]:
    """Verify the ACME account exists, has status='valid', and has an
    account_url stored.
    """
    started = time.time()
    if not account_id:
        return _check_result(
            "account",
            "ACME account",
            "fail",
            "Order has no ACME account id",
            severity="error",
            duration_ms=int((time.time() - started) * 1000),
        )
    row = await conn.fetchrow(
        "SELECT id, email, status, account_url FROM letsencrypt_accounts WHERE id = $1",
        account_id,
    )
    duration_ms = int((time.time() - started) * 1000)
    if not row:
        return _check_result(
            "account",
            "ACME account",
            "fail",
            f"Account {account_id} not found",
            severity="error",
            duration_ms=duration_ms,
        )
    if row["status"] != "valid":
        return _check_result(
            "account",
            "ACME account",
            "fail",
            f"Account status is '{row['status']}', expected 'valid'",
            severity="error",
            details={"account": dict(row)},
            duration_ms=duration_ms,
        )
    if not row["account_url"]:
        return _check_result(
            "account",
            "ACME account",
            "warn",
            "Account has no account_url stored",
            severity="warn",
            details={"account": dict(row)},
            duration_ms=duration_ms,
        )
    return _check_result(
        "account",
        "ACME account",
        "ok",
        f"Account {row['email']} is valid",
        severity="info",
        details={"account": dict(row)},
        duration_ms=duration_ms,
    )


async def check_agents(conn, cluster_ids: List[int]) -> Dict[str, Any]:
    """Verify at least one healthy agent is registered for the order's
    cluster(s).
    """
    started = time.time()
    if not cluster_ids:
        return _check_result(
            "agents",
            "HAProxy agents",
            "warn",
            "Order has no associated cluster",
            severity="warn",
            duration_ms=int((time.time() - started) * 1000),
        )
    # Bulgu #84 (round-23 audit) — the canonical timestamp column on the
    # `agents` table is `last_seen`. Pre-fix this query referenced a
    # non-existent `a.last_heartbeat`, so every ACME preflight call
    # (`POST /api/sites/preflight-acme`) crashed at the `check_agents`
    # stage with `UndefinedColumnError: column a.last_heartbeat does
    # not exist`, blocking the entire wizard's ACME pre-validation
    # gate. Every other agents.last_seen reader in the codebase
    # (routers/cluster.py:695-702, routers/agent.py, routers/dashboard
    # *.py) uses `last_seen`; aligning here.
    rows = await conn.fetch(
        """
        SELECT a.id, a.hostname, a.status, a.last_seen, hc.id AS cluster_id, hc.name AS cluster_name
        FROM agents a
        JOIN haproxy_clusters hc ON hc.pool_id = a.pool_id
        WHERE hc.id = ANY($1::int[])
        """,
        cluster_ids,
    )
    duration_ms = int((time.time() - started) * 1000)
    if not rows:
        return _check_result(
            "agents",
            "HAProxy agents",
            "fail",
            "No agents registered for the target cluster(s)",
            severity="error",
            details={"cluster_ids": cluster_ids},
            duration_ms=duration_ms,
        )
    healthy = [r for r in rows if r["status"] in ("active", "online")]
    if not healthy:
        return _check_result(
            "agents",
            "HAProxy agents",
            "warn",
            f"{len(rows)} agent(s) registered but none currently active",
            severity="warn",
            details={"agents": [dict(r) for r in rows]},
            duration_ms=duration_ms,
        )
    return _check_result(
        "agents",
        "HAProxy agents",
        "ok",
        f"{len(healthy)} of {len(rows)} agents are active",
        severity="info",
        details={"agents": [dict(r) for r in rows]},
        duration_ms=duration_ms,
    )


# ----------------------------------------------------------------------------
# Public orchestration
# ----------------------------------------------------------------------------


CHECK_IDS = ("dns", "port80", "routing", "account", "agents")


def _coerce_cluster_ids(raw) -> List[int]:
    """Coerce a cluster_ids list to ints, dropping non-integer-compatible
    values. JSONB-stored lists occasionally land as ["1", "2"] (string form)
    due to legacy paths; asyncpg's `$1::int[]` cast then fails the diagnostic
    query with InvalidTextRepresentationError. We normalise here so the
    diagnostic surface is the same regardless of how the order was written.
    """
    out: List[int] = []
    if not raw:
        return out
    for v in raw:
        try:
            out.append(int(v))
        except (TypeError, ValueError):
            continue
    return out


# Bulgu #94 (Round-25 audit) — the entire point of the diagnostic panel
# is to SHOW the operator what went wrong. Pre-fix, a single check raising
# an uncaught exception (e.g. an asyncpg cast error from a malformed
# cluster_ids JSONB, a DNS resolver outage, an SSRF-guard glitch) would
# propagate up to the router's `try/finally` block, which had no `except`
# clause, and return HTTP 500 with no body. The operator saw only
# "Internal Server Error" in DevTools — the inverse of what a diagnostic
# panel should ever produce. We now wrap every check inside `run_checks`
# so that a check crash becomes a structured `fail` row instead of
# bubbling up; the operator gets the exception type + message in the
# UI and can carry it forward, and the rest of the panel still renders.
async def _safe_check(check_id: str, label: str, coro):
    """Run an awaitable that produces a check result; swallow exceptions
    and convert them to a structured `fail` result so the diagnostic
    response is never short-circuited by a single broken check."""
    started = time.time()
    try:
        return await coro
    except Exception as exc:  # noqa: BLE001 — diagnostic boundary
        duration_ms = int((time.time() - started) * 1000)
        logger.exception(
            "ACME diagnostic check %s raised", check_id
        )
        return _check_result(
            check_id,
            label,
            "fail",
            f"Diagnostic check crashed: {exc.__class__.__name__}: {exc}",
            severity="error",
            details={
                "exception_type": exc.__class__.__name__,
                "exception_message": str(exc),
            },
            duration_ms=duration_ms,
        )


async def run_checks(
    conn,
    *,
    domains: List[str],
    cluster_ids: List[int],
    account_id: Optional[int],
    only: Optional[List[str]] = None,
    challenge_type: str = "http-01",
) -> List[Dict[str, Any]]:
    """Execute the full pre-flight check suite. `only` lets callers re-run a
    subset (per-check rerun in the UI).

    Every individual check is wrapped in `_safe_check` so the diagnostic
    endpoint NEVER 500s because of one broken check — the operator gets
    a structured `fail` row identifying which check crashed and why.
    """
    selected = set(only) if only else set(CHECK_IDS)
    results: List[Dict[str, Any]] = []

    # Normalise inputs once so the per-check error stays in the right
    # bucket (a malformed cluster_ids should not crash routing/agents).
    safe_domains = [d for d in (domains or []) if isinstance(d, str) and d]
    safe_cluster_ids = _coerce_cluster_ids(cluster_ids)
    try:
        safe_account_id = int(account_id) if account_id is not None else None
    except (TypeError, ValueError):
        safe_account_id = None

    # Issue #35: DNS-01 validates via a TXT record, so the HTTP-01 reachability checks
    # (public A record, inbound port 80, ACME Challenge Routing) do not apply — report them
    # as `skipped` rather than failing an internal/isolated host that is actually fine.
    is_dns01 = (challenge_type == "dns-01")
    if "dns" in selected:
        if is_dns01:
            results.append(_check_result("dns", "DNS resolution", "skipped",
                                         "DNS-01: a public A record is not required (validation is via a TXT record).",
                                         severity="info"))
        else:
            results.append(await _safe_check("dns", "DNS resolution", check_dns(safe_domains)))
    if "port80" in selected:
        if is_dns01:
            results.append(_check_result("port80", "Port 80 reachability", "skipped",
                                         "DNS-01: inbound port 80 is not required.", severity="info"))
        else:
            results.append(await _safe_check("port80", "Port 80 reachability", check_port80(safe_domains)))
    if "routing" in selected:
        if is_dns01:
            results.append(_check_result("routing", "HAProxy routing", "skipped",
                                         "DNS-01: ACME Challenge Routing is not required.", severity="info"))
        else:
            results.append(await _safe_check("routing", "HAProxy routing", check_routing(conn, safe_domains, safe_cluster_ids)))
    if "account" in selected:
        results.append(await _safe_check("account", "ACME account", check_account(conn, safe_account_id)))
    if "agents" in selected:
        results.append(await _safe_check("agents", "HAProxy agents", check_agents(conn, safe_cluster_ids)))

    return results