feat(reboot): verify public maintenance fallback

scripts/reboot-recovery/public-maintenance-fallback-probe.py

@@ -0,0 +1,213 @@
+#!/usr/bin/env python3
+"""Probe public routes for raw 5xx exposure without a maintenance fallback."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import socket
+import sys
+import urllib.error
+import urllib.request
+from datetime import datetime
+from pathlib import Path
+from typing import Any
+
+
+SCHEMA_VERSION = "awoooi_public_maintenance_fallback_probe_v1"
+DEFAULT_URLS = [
+    "https://awoooi.wooo.work/api/v1/health",
+    "https://awoooi.wooo.work/",
+    "https://stock.wooo.work/api/v1/system/freshness",
+    "https://mo.wooo.work/health",
+    "https://bitan.wooo.work/",
+    "https://www.tsenyang.com/",
+]
+MAINTENANCE_MARKERS = [
+    "AWOOOI 維護模式",
+    "系統正在恢復服務",
+    "HTTP 502 / 503 / 504 fallback page",
+]
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Read-only public route verifier for maintenance fallback.",
+    )
+    parser.add_argument(
+        "--url",
+        action="append",
+        dest="urls",
+        help="Public URL to probe. May be passed more than once.",
+    )
+    parser.add_argument("--timeout", type=float, default=8.0)
+    parser.add_argument("--max-bytes", type=int, default=65536)
+    parser.add_argument("--generated-at", help="Override generated_at for snapshots.")
+    parser.add_argument("--output", type=Path, help="Write JSON report to this path.")
+    parser.add_argument(
+        "--prometheus-output",
+        type=Path,
+        help="Optionally write node_exporter textfile metrics.",
+    )
+    return parser.parse_args()
+
+
+def fetch_url(url: str, *, timeout: float, max_bytes: int) -> dict[str, Any]:
+    request = urllib.request.Request(
+        url,
+        headers={
+            "User-Agent": "awoooi-reboot-public-maintenance-probe/1.0",
+            "Accept": "text/html,application/json;q=0.9,*/*;q=0.1",
+        },
+    )
+    status = 0
+    headers: dict[str, str] = {}
+    body = b""
+    error = ""
+    try:
+        with urllib.request.urlopen(request, timeout=timeout) as response:
+            status = int(response.getcode() or 0)
+            headers = {key.lower(): value for key, value in response.headers.items()}
+            body = response.read(max_bytes)
+    except urllib.error.HTTPError as exc:
+        status = int(exc.code or 0)
+        headers = {key.lower(): value for key, value in exc.headers.items()}
+        try:
+            body = exc.read(max_bytes)
+        except OSError:
+            body = b""
+    except (urllib.error.URLError, TimeoutError, socket.timeout) as exc:
+        error = str(getattr(exc, "reason", exc))
+
+    text = body.decode("utf-8", errors="replace")
+    fallback_header = headers.get("x-awoooi-fallback", "")
+    maintenance_body_detected = any(marker in text for marker in MAINTENANCE_MARKERS)
+    maintenance_fallback_serving = (
+        status in {200, 503}
+        and fallback_header == "local-maintenance"
+        and maintenance_body_detected
+    )
+    raw_5xx_without_fallback = (
+        status >= 500
+        and not maintenance_fallback_serving
+    )
+    route_unreachable_without_external_fallback = status == 0
+    if maintenance_fallback_serving:
+        route_status = "maintenance_fallback_serving"
+    elif raw_5xx_without_fallback:
+        route_status = "raw_5xx_without_maintenance_fallback"
+    elif route_unreachable_without_external_fallback:
+        route_status = "route_unreachable_without_external_fallback"
+    elif 200 <= status < 500:
+        route_status = "normal_origin_serving"
+    else:
+        route_status = "unknown"
+
+    return {
+        "url": url,
+        "http_status": status,
+        "status": route_status,
+        "fallback_header": fallback_header,
+        "maintenance_body_detected": maintenance_body_detected,
+        "maintenance_fallback_serving": maintenance_fallback_serving,
+        "raw_5xx_without_fallback": raw_5xx_without_fallback,
+        "route_unreachable_without_external_fallback": (
+            route_unreachable_without_external_fallback
+        ),
+        "error": error[:300],
+    }
+
+
+def build_report(args: argparse.Namespace) -> dict[str, Any]:
+    generated_at = args.generated_at or datetime.now().astimezone().isoformat(
+        timespec="seconds"
+    )
+    urls = args.urls or DEFAULT_URLS
+    routes = [
+        fetch_url(url, timeout=args.timeout, max_bytes=args.max_bytes) for url in urls
+    ]
+    raw_5xx_without_fallback = [
+        row["url"] for row in routes if row["raw_5xx_without_fallback"]
+    ]
+    unreachable_without_external_fallback = [
+        row["url"]
+        for row in routes
+        if row["route_unreachable_without_external_fallback"]
+    ]
+    maintenance_fallback_routes = [
+        row["url"] for row in routes if row["maintenance_fallback_serving"]
+    ]
+    blockers: list[str] = []
+    if raw_5xx_without_fallback:
+        blockers.append("public_route_raw_5xx_without_maintenance_fallback")
+    if unreachable_without_external_fallback:
+        blockers.append("public_route_unreachable_without_external_l1_fallback")
+    return {
+        "schema_version": SCHEMA_VERSION,
+        "generated_at": generated_at,
+        "runtime_readback_present": True,
+        "ready": not blockers,
+        "target_url_count": len(urls),
+        "route_count": len(routes),
+        "raw_5xx_without_fallback_count": len(raw_5xx_without_fallback),
+        "route_unreachable_without_external_fallback_count": len(
+            unreachable_without_external_fallback
+        ),
+        "maintenance_fallback_route_count": len(maintenance_fallback_routes),
+        "raw_5xx_without_fallback_urls": raw_5xx_without_fallback,
+        "route_unreachable_without_external_fallback_urls": (
+            unreachable_without_external_fallback
+        ),
+        "maintenance_fallback_urls": maintenance_fallback_routes,
+        "blockers": blockers,
+        "routes": routes,
+        "forbidden_actions": [
+            "nginx_reload_from_probe",
+            "dns_cutover_from_probe",
+            "service_restart_from_probe",
+            "host_reboot_from_probe",
+            "secret_value_read",
+        ],
+    }
+
+
+def write_prometheus(path: Path, report: dict[str, Any]) -> None:
+    ready = 1 if report.get("ready") is True else 0
+    raw_5xx = int(report.get("raw_5xx_without_fallback_count") or 0)
+    unreachable = int(
+        report.get("route_unreachable_without_external_fallback_count") or 0
+    )
+    text = "\n".join(
+        [
+            "# HELP awoooi_public_maintenance_fallback_ready Whether public routes avoid raw 5xx exposure.",
+            "# TYPE awoooi_public_maintenance_fallback_ready gauge",
+            f'awoooi_public_maintenance_fallback_ready{{scope="public_routes"}} {ready}',
+            "# HELP awoooi_public_route_raw_5xx_without_maintenance_fallback Raw public 5xx responses without maintenance fallback.",
+            "# TYPE awoooi_public_route_raw_5xx_without_maintenance_fallback gauge",
+            f'awoooi_public_route_raw_5xx_without_maintenance_fallback{{scope="public_routes"}} {raw_5xx}',
+            "# HELP awoooi_public_route_unreachable_without_external_l1_fallback Public routes unreachable without external fallback proof.",
+            "# TYPE awoooi_public_route_unreachable_without_external_l1_fallback gauge",
+            f'awoooi_public_route_unreachable_without_external_l1_fallback{{scope="public_routes"}} {unreachable}',
+            "",
+        ]
+    )
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(text, encoding="utf-8")
+
+
+def main() -> int:
+    args = parse_args()
+    report = build_report(args)
+    text = json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+    if args.output:
+        args.output.parent.mkdir(parents=True, exist_ok=True)
+        args.output.write_text(text, encoding="utf-8")
+    else:
+        print(text, end="")
+    if args.prometheus_output:
+        write_prometheus(args.prometheus_output, report)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/reboot-recovery/reboot-auto-recovery-slo-exporter.sh

@@ -16,6 +16,8 @@ LOCK_FILE="${LOCK_FILE:-${LOG_DIR}/reboot_auto_recovery_slo.lock}"
 STOCK_FRESHNESS_URL="${STOCK_FRESHNESS_URL:-https://stock.wooo.work/api/v1/system/freshness}"
 STOCK_INGESTION_URL="${STOCK_INGESTION_URL:-https://stock.wooo.work/api/v1/system/ingestion}"
 STOCK_READBACK_TIMEOUT_SECONDS="${STOCK_READBACK_TIMEOUT_SECONDS:-10}"
+PUBLIC_MAINTENANCE_READBACK_TIMEOUT_SECONDS="${PUBLIC_MAINTENANCE_READBACK_TIMEOUT_SECONDS:-8}"
+PUBLIC_MAINTENANCE_URLS="${PUBLIC_MAINTENANCE_URLS:-https://awoooi.wooo.work/api/v1/health https://awoooi.wooo.work/ https://stock.wooo.work/api/v1/system/freshness https://mo.wooo.work/health https://bitan.wooo.work/ https://www.tsenyang.com/}"
 
 mkdir -p "$TEXTFILE_DIR" "$LOG_DIR"
 
@@ -35,6 +37,8 @@ summary_file="$artifact_dir/summary.txt"
 scorecard_file="$artifact_dir/scorecard.json"
 stock_freshness_file="$artifact_dir/stock-freshness.json"
 stock_ingestion_file="$artifact_dir/stock-ingestion.json"
+public_maintenance_file="$artifact_dir/public-maintenance-fallback.json"
+public_maintenance_prom="$artifact_dir/public-maintenance-fallback.prom"
 windows99_management_file="$artifact_dir/windows99-management-channel.json"
 reboot_event_state_file="${REBOOT_EVENT_STATE_FILE:-${LOG_DIR}/reboot-event-state.json}"
 
@@ -57,6 +61,20 @@ if command -v curl >/dev/null 2>&1; then
     || rm -f "$stock_ingestion_file"
 fi
 
+if [ -f "$ROOT_DIR/scripts/reboot-recovery/public-maintenance-fallback-probe.py" ]; then
+  public_probe_args=(
+    "$ROOT_DIR/scripts/reboot-recovery/public-maintenance-fallback-probe.py"
+    --timeout "$PUBLIC_MAINTENANCE_READBACK_TIMEOUT_SECONDS"
+    --output "$public_maintenance_file"
+    --prometheus-output "$public_maintenance_prom"
+  )
+  for url in $PUBLIC_MAINTENANCE_URLS; do
+    public_probe_args+=(--url "$url")
+  done
+  python3 "${public_probe_args[@]}" >"$artifact_dir/public-maintenance-fallback.stdout" 2>"$artifact_dir/public-maintenance-fallback.err" \
+    || rm -f "$public_maintenance_file" "$public_maintenance_prom"
+fi
+
 if [ -f "$ROOT_DIR/scripts/reboot-recovery/windows99-management-channel-probe.py" ]; then
   python3 "$ROOT_DIR/scripts/reboot-recovery/windows99-management-channel-probe.py" \
     --output "$windows99_management_file" >"$artifact_dir/windows99-management-channel.stdout" 2>"$artifact_dir/windows99-management-channel.err" \
@@ -79,6 +97,9 @@ fi
 if [ -s "$stock_ingestion_file" ]; then
   scorecard_args+=(--stock-ingestion-file "$stock_ingestion_file")
 fi
+if [ -s "$public_maintenance_file" ]; then
+  scorecard_args+=(--public-maintenance-file "$public_maintenance_file")
+fi
 if [ -s "$windows99_management_file" ]; then
   scorecard_args+=(--windows99-management-file "$windows99_management_file")
 fi
@@ -110,6 +131,24 @@ payload=json.load(open(sys.argv[1], encoding="utf-8"))
 print(payload.get("sla_recovery_eta", {}).get("target_seconds_remaining", 0) or 0)
 PY
 )"
+public_maintenance_ready="$(python3 - "$scorecard_file" <<'PY'
+import json, sys
+payload=json.load(open(sys.argv[1], encoding="utf-8"))
+print(1 if payload.get("public_maintenance_fallback", {}).get("ready") else 0)
+PY
+)"
+public_raw_5xx_count="$(python3 - "$scorecard_file" <<'PY'
+import json, sys
+payload=json.load(open(sys.argv[1], encoding="utf-8"))
+print(payload.get("public_maintenance_fallback", {}).get("raw_5xx_without_fallback_count", 0) or 0)
+PY
+)"
+public_unreachable_count="$(python3 - "$scorecard_file" <<'PY'
+import json, sys
+payload=json.load(open(sys.argv[1], encoding="utf-8"))
+print(payload.get("public_maintenance_fallback", {}).get("route_unreachable_without_external_fallback_count", 0) or 0)
+PY
+)"
 
 tmp_metric="$(mktemp "$TEXTFILE_DIR/.reboot_auto_recovery_slo.XXXXXX")"
 cat >"$tmp_metric" <<METRICS
@@ -128,6 +167,15 @@ awoooi_reboot_auto_recovery_slo_target_seconds_remaining{scope="99_110_111_112_1
 # HELP awoooi_reboot_auto_recovery_slo_last_run_timestamp Last SLO exporter run timestamp.
 # TYPE awoooi_reboot_auto_recovery_slo_last_run_timestamp gauge
 awoooi_reboot_auto_recovery_slo_last_run_timestamp{scope="99_110_111_112_120_121_188"} $now
+# HELP awoooi_public_maintenance_fallback_ready Whether public routes avoid raw 5xx exposure.
+# TYPE awoooi_public_maintenance_fallback_ready gauge
+awoooi_public_maintenance_fallback_ready{scope="public_routes"} $public_maintenance_ready
+# HELP awoooi_public_route_raw_5xx_without_maintenance_fallback Raw public 5xx responses without maintenance fallback.
+# TYPE awoooi_public_route_raw_5xx_without_maintenance_fallback gauge
+awoooi_public_route_raw_5xx_without_maintenance_fallback{scope="public_routes"} $public_raw_5xx_count
+# HELP awoooi_public_route_unreachable_without_external_l1_fallback Public routes unreachable without external fallback proof.
+# TYPE awoooi_public_route_unreachable_without_external_l1_fallback gauge
+awoooi_public_route_unreachable_without_external_l1_fallback{scope="public_routes"} $public_unreachable_count
 METRICS
 if [ -s "$reboot_event_prom" ]; then
   cat "$reboot_event_prom" >>"$tmp_metric"

scripts/reboot-recovery/reboot-auto-recovery-slo-scorecard.py

@@ -44,6 +44,11 @@ def parse_args() -> argparse.Namespace:
         type=Path,
         help="Optional host pressure JSON readback from Prometheus / node exporter.",
     )
+    parser.add_argument(
+        "--public-maintenance-file",
+        type=Path,
+        help="Optional public maintenance fallback route probe JSON readback.",
+    )
     parser.add_argument(
         "--windows99-vmware-file",
         type=Path,
@@ -347,6 +352,93 @@ def read_json_object(path: Path | None) -> dict[str, Any]:
     return payload
 
 
+def parse_public_maintenance_readback(path: Path | None) -> dict[str, Any]:
+    """Parse public route maintenance fallback probe JSON."""
+    default = {
+        "runtime_readback_present": False,
+        "ready": False,
+        "target_url_count": 0,
+        "route_count": 0,
+        "raw_5xx_without_fallback_count": 0,
+        "route_unreachable_without_external_fallback_count": 0,
+        "maintenance_fallback_route_count": 0,
+        "raw_5xx_without_fallback_urls": [],
+        "route_unreachable_without_external_fallback_urls": [],
+        "maintenance_fallback_urls": [],
+        "routes": [],
+        "blockers": ["public_maintenance_fallback_runtime_readback_missing"],
+    }
+    payload = read_json_object(path)
+    if not payload:
+        return default
+
+    routes = payload.get("routes")
+    if not isinstance(routes, list):
+        routes = []
+    compact_routes: list[dict[str, Any]] = []
+    for item in routes:
+        if not isinstance(item, dict):
+            continue
+        compact_routes.append(
+            {
+                "url": str(item.get("url") or ""),
+                "http_status": int_value(item.get("http_status")),
+                "status": str(item.get("status") or "unknown"),
+                "fallback_header": str(item.get("fallback_header") or ""),
+                "maintenance_body_detected": (
+                    item.get("maintenance_body_detected") is True
+                ),
+                "maintenance_fallback_serving": (
+                    item.get("maintenance_fallback_serving") is True
+                ),
+                "raw_5xx_without_fallback": (
+                    item.get("raw_5xx_without_fallback") is True
+                ),
+                "route_unreachable_without_external_fallback": (
+                    item.get("route_unreachable_without_external_fallback") is True
+                ),
+                "error": str(item.get("error") or "")[:300],
+            }
+        )
+
+    blockers = strings(payload.get("blockers"))
+    raw_5xx_count = int_value(payload.get("raw_5xx_without_fallback_count"))
+    unreachable_count = int_value(
+        payload.get("route_unreachable_without_external_fallback_count")
+    )
+    if raw_5xx_count and "public_route_raw_5xx_without_maintenance_fallback" not in blockers:
+        blockers.append("public_route_raw_5xx_without_maintenance_fallback")
+    if (
+        unreachable_count
+        and "public_route_unreachable_without_external_l1_fallback" not in blockers
+    ):
+        blockers.append("public_route_unreachable_without_external_l1_fallback")
+
+    return {
+        "runtime_readback_present": True,
+        "schema_version": str(payload.get("schema_version") or "unknown"),
+        "generated_at": str(payload.get("generated_at") or ""),
+        "ready": payload.get("ready") is True and not blockers,
+        "target_url_count": int_value(payload.get("target_url_count")),
+        "route_count": int_value(payload.get("route_count")) or len(compact_routes),
+        "raw_5xx_without_fallback_count": raw_5xx_count,
+        "route_unreachable_without_external_fallback_count": unreachable_count,
+        "maintenance_fallback_route_count": int_value(
+            payload.get("maintenance_fallback_route_count")
+        ),
+        "raw_5xx_without_fallback_urls": strings(
+            payload.get("raw_5xx_without_fallback_urls")
+        ),
+        "route_unreachable_without_external_fallback_urls": strings(
+            payload.get("route_unreachable_without_external_fallback_urls")
+        ),
+        "maintenance_fallback_urls": strings(payload.get("maintenance_fallback_urls")),
+        "routes": compact_routes,
+        "blockers": blockers,
+        "forbidden_actions": strings(payload.get("forbidden_actions")),
+    }
+
+
 def parse_host_probe(text: str) -> list[dict[str, Any]]:
     rows: list[dict[str, Any]] = []
     for raw_line in text.splitlines():
@@ -446,6 +538,12 @@ def source_controls() -> dict[str, bool]:
         "public_maintenance_fallback_source_present": source_file(
             "ops/maintenance/maintenance.html"
         ).exists()
+        and file_contains(
+            source_file("ops/maintenance/nginx-502-maintenance-snippet.conf"),
+            "proxy_intercept_errors on",
+            "error_page 502 503 504",
+            "X-AWOOOI-Fallback",
+        )
         and file_contains(
             source_file("docs/runbooks/PUBLIC-MAINTENANCE-FALLBACK-RUNBOOK.md"),
             "L0",
@@ -886,6 +984,7 @@ def choose_safe_next_step(
     blockers: list[str],
     stockplatform: dict[str, Any],
     host_pressure: dict[str, Any],
+    public_maintenance: dict[str, Any],
 ) -> str:
     freshness_status = str(stockplatform.get("freshness_status") or "unknown")
     eod_window = stockplatform.get("eod_window") if isinstance(stockplatform.get("eod_window"), dict) else {}
@@ -934,6 +1033,22 @@ def choose_safe_next_step(
             "collect_windows99_vmware_autostart_verify_readback_then_rerun_all_host_"
             "reboot_scorecard_no_secret_no_reboot"
         )
+    public_blockers = set(strings(public_maintenance.get("blockers")))
+    if "public_route_raw_5xx_without_maintenance_fallback" in public_blockers:
+        return (
+            "apply_public_maintenance_fallback_l0_nginx_intercept_with_nginx_t_"
+            "test_staging_vhost_header_verifier_then_deploy_no_app_restart"
+        )
+    if "public_route_unreachable_without_external_l1_fallback" in public_blockers:
+        return (
+            "prepare_public_maintenance_fallback_l1_external_static_origin_or_cdn_"
+            "healthcheck_decision_record_then_controlled_cutover_with_rollback"
+        )
+    if "public_maintenance_fallback_runtime_readback_missing" in blockers:
+        return (
+            "run_public_maintenance_fallback_probe_and_rerun_reboot_scorecard_"
+            "verify_only_no_nginx_reload"
+        )
     if blockers == ["host_boot_observation_older_than_target_window"]:
         return (
             "timer_deployed_and_services_readback_green_wait_for_next_all_host_reboot_"
@@ -963,6 +1078,9 @@ def build_required_checks(payload: dict[str, Any]) -> dict[str, bool]:
     stockplatform = payload.get("stockplatform_data_freshness")
     if not isinstance(stockplatform, dict):
         stockplatform = {}
+    public_maintenance = payload.get("public_maintenance_fallback")
+    if not isinstance(public_maintenance, dict):
+        public_maintenance = {}
     windows99 = payload.get("windows99_vmware_autostart")
     if not isinstance(windows99, dict):
         windows99 = {}
@@ -1001,6 +1119,9 @@ def build_required_checks(payload: dict[str, Any]) -> dict[str, bool]:
         "stockplatform_ingestion_ok": (
             stockplatform.get("ingestion_status") in {"ok", "unknown"}
         ),
+        "public_maintenance_fallback_runtime_ready": (
+            public_maintenance.get("ready") is True
+        ),
         "fresh_reboot_window_observed": not strings(
             host_boot_detection.get("stale_hosts")
         ),
@@ -1042,6 +1163,8 @@ def reboot_sop_current_phase(active_blockers: list[str], can_claim: bool) -> str
         "host_188_service_green_not_1",
         "wazuh_dashboard_degraded",
     }
+    if any(blocker.startswith("public_") for blocker in active_blockers):
+        return "public_maintenance_fallback_blocked"
     if any(blocker in service_blockers for blocker in active_blockers):
         return "post_reboot_service_readiness_blocked"
     if any("stockplatform" in blocker or "product_data" in blocker for blocker in active_blockers):
@@ -1072,6 +1195,9 @@ def reboot_sop_primary_blocker(active_blockers: list[str]) -> str:
         "windows99_vmware_autostart_config_not_ready",
         "windows99_vmware_guest_power_not_ready",
         "windows99_update_no_auto_reboot_policy_not_ready",
+        "public_route_raw_5xx_without_maintenance_fallback",
+        "public_route_unreachable_without_external_l1_fallback",
+        "public_maintenance_fallback_runtime_readback_missing",
         "post_start_blocked_not_zero",
         "service_green_not_1",
         "host_188_service_green_not_1",
@@ -1159,6 +1285,9 @@ def enrich_machine_readback(payload: dict[str, Any]) -> dict[str, Any]:
     stockplatform = payload.get("stockplatform_data_freshness")
     if not isinstance(stockplatform, dict):
         stockplatform = {}
+    public_maintenance = payload.get("public_maintenance_fallback")
+    if not isinstance(public_maintenance, dict):
+        public_maintenance = {}
     windows99 = payload.get("windows99_vmware_autostart")
     if not isinstance(windows99, dict):
         windows99 = {}
@@ -1253,6 +1382,23 @@ def enrich_machine_readback(payload: dict[str, Any]) -> dict[str, Any]:
         "stockplatform_ingestion_blocker_count": len(
             strings(stockplatform.get("ingestion_blockers"))
         ),
+        "public_maintenance_runtime_readback_present": (
+            public_maintenance.get("runtime_readback_present") is True
+        ),
+        "public_maintenance_fallback_ready": (
+            public_maintenance.get("ready") is True
+        ),
+        "public_route_raw_5xx_without_fallback_count": int_value(
+            public_maintenance.get("raw_5xx_without_fallback_count")
+        ),
+        "public_route_unreachable_without_external_fallback_count": int_value(
+            public_maintenance.get(
+                "route_unreachable_without_external_fallback_count"
+            )
+        ),
+        "public_maintenance_fallback_route_count": int_value(
+            public_maintenance.get("maintenance_fallback_route_count")
+        ),
         "windows99_vmware_readback_present": windows99.get("readback_present") is True,
         "windows99_vmrun_present": windows99.get("vmrun_present") is True,
         "windows99_vmware_config_ready": windows99.get("config_ready") is True,
@@ -1338,6 +1484,18 @@ def enrich_machine_readback(payload: dict[str, Any]) -> dict[str, Any]:
             "windows99_remote_execution_channel_ready"
         ],
         "windows99_ssh_batch_status": rollups["windows99_ssh_batch_status"],
+        "public_maintenance_runtime_readback_present": rollups[
+            "public_maintenance_runtime_readback_present"
+        ],
+        "public_maintenance_fallback_ready": rollups[
+            "public_maintenance_fallback_ready"
+        ],
+        "public_route_raw_5xx_without_fallback_count": rollups[
+            "public_route_raw_5xx_without_fallback_count"
+        ],
+        "public_route_unreachable_without_external_fallback_count": rollups[
+            "public_route_unreachable_without_external_fallback_count"
+        ],
         "runtime_write_authorized_by_this_scorecard": False,
         "host_reboot_authorized_by_this_scorecard": False,
         "workflow_trigger_authorized_by_this_scorecard": False,
@@ -1371,6 +1529,15 @@ def enrich_machine_readback(payload: dict[str, Any]) -> dict[str, Any]:
         "reboot_auto_recovery_stockplatform_ingestion_status": rollups[
             "stockplatform_ingestion_status"
         ],
+        "reboot_auto_recovery_public_maintenance_fallback_ready": rollups[
+            "public_maintenance_fallback_ready"
+        ],
+        "reboot_auto_recovery_public_route_raw_5xx_without_fallback_count": rollups[
+            "public_route_raw_5xx_without_fallback_count"
+        ],
+        "reboot_auto_recovery_public_route_unreachable_without_external_fallback_count": (
+            rollups["public_route_unreachable_without_external_fallback_count"]
+        ),
         "reboot_auto_recovery_safe_next_step": readback["safe_next_step"],
         "reboot_auto_recovery_next_safe_action": readback["next_safe_action"],
         "reboot_auto_recovery_source_controls_present": source_controls_present,
@@ -1419,6 +1586,9 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
         generated_at=generated_at,
     )
     host_pressure = build_host_pressure_readback(read_json_object(args.host_pressure_file))
+    public_maintenance = parse_public_maintenance_readback(
+        args.public_maintenance_file
+    )
     windows99 = parse_windows99_vmware_readback(read_text(args.windows99_vmware_file))
     windows99_management = parse_windows99_management_readback(
         args.windows99_management_file
@@ -1503,6 +1673,7 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
     if free_gib is not None and free_gib < args.min_free_gib:
         blockers.append("local_disk_free_below_minimum")
     blockers.extend(strings(host_pressure.get("blockers")))
+    blockers.extend(strings(public_maintenance.get("blockers")))
     blockers.extend(strings(windows99.get("blockers")))
     if (
         windows99.get("readback_present") is False
@@ -1521,6 +1692,7 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
         blockers=unique_blockers,
         stockplatform=stockplatform,
         host_pressure=host_pressure,
+        public_maintenance=public_maintenance,
     )
     payload = {
         "schema_version": SCHEMA_VERSION,
@@ -1591,6 +1763,7 @@ def build_scorecard(args: argparse.Namespace) -> dict[str, Any]:
         },
         "stockplatform_data_freshness": stockplatform,
         "host_pressure": host_pressure,
+        "public_maintenance_fallback": public_maintenance,
         "windows99_vmware_autostart": windows99,
         "windows99_management_channel": windows99_management,
         "capacity": {

scripts/reboot-recovery/tests/test_reboot_p0_operational_contract.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import json
+import subprocess
+import sys
 from pathlib import Path
 
 
@@ -38,6 +41,8 @@ def test_reboot_p0_contract_has_maintenance_and_telegram_alerts() -> None:
     alerts = read("ops/monitoring/alerts-unified.yml")
     runbook = read("docs/runbooks/PUBLIC-MAINTENANCE-FALLBACK-RUNBOOK.md")
     snippet = read("ops/maintenance/nginx-502-maintenance-snippet.conf")
+    probe = read("scripts/reboot-recovery/public-maintenance-fallback-probe.py")
+    exporter = read("scripts/reboot-recovery/reboot-auto-recovery-slo-exporter.sh")
 
     for alert_name in [
         "HostRebootEventDetected",
@@ -48,9 +53,131 @@ def test_reboot_p0_contract_has_maintenance_and_telegram_alerts() -> None:
         assert alert_name in alerts
     assert "notification_type: TYPE-3" in alerts
     assert "proxy_intercept_errors on" in snippet
+    assert "X-AWOOOI-Fallback" in snippet
     assert "502" in runbook
     assert "L0" in runbook
     assert "L1" in runbook
+    assert "raw_5xx_without_fallback" in probe
+    assert "public-maintenance-fallback.json" in exporter
+    assert "--public-maintenance-file" in exporter
+    assert "awoooi_public_route_raw_5xx_without_maintenance_fallback" in exporter
+
+
+def test_reboot_slo_scorecard_fails_closed_on_raw_public_502(tmp_path: Path) -> None:
+    summary = tmp_path / "summary.txt"
+    summary.write_text(
+        "\n".join(
+            [
+                "SERVICE_GREEN=1",
+                "PRODUCT_DATA_GREEN=1",
+                "BACKUP_CORE_GREEN=1",
+                "POST_START_BLOCKED=0",
+                "HOST_188_SERVICE_GREEN=1",
+            ]
+        ),
+        encoding="utf-8",
+    )
+    host_probe = tmp_path / "host-probe.txt"
+    host_probe.write_text(
+        "\n".join(
+            f"HOST_BOOT alias={alias} host=192.168.0.{alias} reachable=1 uptime_seconds=120"
+            for alias in ["99", "110", "111", "112", "120", "121", "188"]
+        ),
+        encoding="utf-8",
+    )
+    reboot_event = tmp_path / "reboot-event.json"
+    reboot_event.write_text(
+        json.dumps(
+            {
+                "reboot_detected": True,
+                "all_required_hosts_observed": True,
+                "all_required_hosts_in_reboot_window": True,
+                "fresh_boot_hosts": ["99", "110", "111", "112", "120", "121", "188"],
+                "missing_hosts": [],
+                "unreachable_hosts": [],
+                "target_seconds_remaining": 480,
+                "recovery_deadline_status": "inside_target_window",
+            }
+        ),
+        encoding="utf-8",
+    )
+    public_probe = tmp_path / "public-maintenance.json"
+    public_probe.write_text(
+        json.dumps(
+            {
+                "runtime_readback_present": True,
+                "ready": False,
+                "raw_5xx_without_fallback_count": 1,
+                "route_unreachable_without_external_fallback_count": 0,
+                "maintenance_fallback_route_count": 0,
+                "raw_5xx_without_fallback_urls": ["https://awoooi.wooo.work/"],
+                "blockers": ["public_route_raw_5xx_without_maintenance_fallback"],
+                "routes": [
+                    {
+                        "url": "https://awoooi.wooo.work/",
+                        "http_status": 502,
+                        "status": "raw_5xx_without_maintenance_fallback",
+                        "fallback_header": "",
+                        "maintenance_body_detected": False,
+                        "maintenance_fallback_serving": False,
+                        "raw_5xx_without_fallback": True,
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+    windows99 = tmp_path / "windows99-vmware.txt"
+    windows99.write_text(
+        "\n".join(
+            [
+                "VMRUN_PRESENT=1",
+                "VMWARE_AUTOSTART_CONFIG_READY=1",
+                "VMWARE_AUTOSTART_POWER_READY=1",
+                "WINDOWS_UPDATE_NO_AUTO_REBOOT_READY=1",
+                "VMWARE_AUTOSTART_VERIFY_READY=1",
+                "VMWARE_SERVICE name=VMAuthdService present=1 status=Running startup_type=Automatic ok=1",
+                "VMWARE_SERVICE name=VMnetDHCP present=1 status=Running startup_type=Automatic ok=1",
+                "VMWARE_AUTOSTART_TASK name=AWOOOI-Start-VMware-VMs present=1 enabled=1 state=Ready trigger_count=1 action_count=1 ok=1",
+                "WINDOWS_UPDATE_POLICY name=NoAutoRebootWithLoggedOnUsers value=1 expected=1 ok=1",
+                *[
+                    f"VMX alias={alias} path=D:\\Documents\\Virtual Machines\\{alias}\\{alias}.vmx present=1"
+                    for alias in ["111", "112", "120", "121", "188"]
+                ],
+                *[
+                    f"VM_POWER alias={alias} vmx_present=1 running=1"
+                    for alias in ["111", "112", "120", "121", "188"]
+                ],
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    result = subprocess.run(
+        [
+            sys.executable,
+            str(ROOT / "scripts/reboot-recovery/reboot-auto-recovery-slo-scorecard.py"),
+            "--summary-file",
+            str(summary),
+            "--host-probe-file",
+            str(host_probe),
+            "--reboot-event-file",
+            str(reboot_event),
+            "--public-maintenance-file",
+            str(public_probe),
+            "--windows99-vmware-file",
+            str(windows99),
+        ],
+        text=True,
+        capture_output=True,
+        check=True,
+    )
+    payload = json.loads(result.stdout)
+
+    assert "public_route_raw_5xx_without_maintenance_fallback" in payload["active_blockers"]
+    assert payload["required_checks"]["public_maintenance_fallback_runtime_ready"] is False
+    assert payload["current_phase"] == "public_maintenance_fallback_blocked"
+    assert payload["rollups"]["public_route_raw_5xx_without_fallback_count"] == 1
 
 
 def test_backup_exporter_emits_domain_level_backup_coverage() -> None: