teleo-infrastructure/ops/check_gcp_infra_readiness.py

#!/usr/bin/env python3
"""Read-only GCP infrastructure readiness check for Teleo.

The check intentionally prints only non-secret metadata. It verifies the CI/CD,
artifact, VM backup, bucket, and DB/KB redundancy surfaces that are easy to
accidentally round up to "ready" without live readbacks.
"""

from __future__ import annotations

import json
import os
import subprocess
from dataclasses import dataclass
from pathlib import Path

PROJECT = "teleo-501523"
REGION = "europe-west6"
ZONE = "europe-west6-a"
ARTIFACT_REPOS = ("teleo", "livingip-web")
BACKUP_BUCKETS = ("teleo-501523-prod-backups", "teleo-501523-leoclean-backups")
DISKS = ("teleo-prod-1", "teleo-staging-1")
SNAPSHOT_POLICY = "teleo-daily-7d-snapshots"
KB_HOST = "teleo@77.42.65.182"
TELEO_VMS = {
    "teleo-prod-1": "sa-teleo-prod-vm@teleo-501523.iam.gserviceaccount.com",
    "teleo-staging-1": "sa-teleo-staging-vm@teleo-501523.iam.gserviceaccount.com",
}
TELEO_SSH_RULES = {
    "teleo-prod-allow-ssh-current-ip": "teleo-prod-ssh",
    "teleo-staging-allow-ssh-current-ip": "teleo-staging-ssh",
}
CLOUDBUILD_SERVICE_ACCOUNT = f"sa-teleo-cloudbuild@{PROJECT}.iam.gserviceaccount.com"
CLOUDBUILD_REQUIRED_ROLES = (
    "roles/artifactregistry.writer",
    "roles/logging.logWriter",
    "roles/storage.objectViewer",
)
GITHUB_ARTIFACT_SERVICE_ACCOUNT = f"sa-artifact-builder@{PROJECT}.iam.gserviceaccount.com"
GITHUB_WIF_POOL = "github-actions"
GITHUB_WIF_PROVIDER = "living-ip-github"
GITHUB_REPOSITORY = "living-ip/teleo-infrastructure"
CLOUDSQL_STANDBY_INSTANCE = "teleo-pgvector-standby"
CLOUDSQL_STANDBY_DATABASE = "teleo_kb"
CLOUDSQL_STANDBY_PASSWORD_SECRET = "gcp-teleo-pgvector-standby-postgres-password"
PROOF_ROOT = Path(os.environ.get("TELEO_GCP_PROOF_ROOT", "outputs/gcp-infra-hardening-20260707/proofs"))


@dataclass
class Check:
    name: str
    status: str
    detail: str
    required_tier: str = "T3_live_readonly"
    current_tier: str = "T3_live_readonly"


def run_json(args: list[str]) -> object:
    output = subprocess.check_output(args, text=True)
    return json.loads(output or "null")


def run_text(args: list[str]) -> str:
    return subprocess.check_output(args, text=True, stderr=subprocess.STDOUT).strip()


def safe_check(name: str, fn) -> Check:
    try:
        return fn()
    except subprocess.CalledProcessError as exc:
        return Check(name, "fail", exc.output.strip()[-500:] if exc.output else str(exc))
    except Exception as exc:  # pragma: no cover - defensive operator readback
        return Check(name, "fail", str(exc))


def check_artifact_repositories() -> Check:
    repos = run_json(
        [
            "gcloud",
            "artifacts",
            "repositories",
            "list",
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    names = {repo.get("name", "").split("/")[-1]: repo for repo in repos}
    missing = [repo for repo in ARTIFACT_REPOS if repo not in names]
    if missing:
        return Check("artifact_repositories", "fail", f"missing={missing}")
    weak = []
    for name in ARTIFACT_REPOS:
        repo = names[name]
        immutable = repo.get("dockerConfig", {}).get("immutableTags") is True
        scanning = repo.get("vulnerabilityScanningConfig", {}).get("enablementState") == "SCANNING_ACTIVE"
        if not immutable or not scanning:
            weak.append(f"{name}:immutable={immutable}:scanning={scanning}")
    if weak:
        return Check("artifact_repositories", "fail", " ".join(weak))
    return Check(
        "artifact_repositories",
        "pass",
        "present_immutable_scanning="
        + ",".join(sorted(names[name].get("name", "") for name in ARTIFACT_REPOS)),
    )


def check_cloud_build_contract() -> Check:
    config = Path("cloudbuild.gcp-staging.yaml")
    dockerfile = Path("Dockerfile.gcp-staging")
    smoke = Path("docker/gcp-staging-smoke.sh")
    missing = [str(path) for path in (config, dockerfile, smoke) if not path.exists()]
    if missing:
        return Check("cloud_build_contract", "fail", f"missing={missing}", current_tier="T1_model")

    text = config.read_text()
    required = [
        "smoke-test-image-before-push",
        "${_REGION}-docker.pkg.dev/$PROJECT_ID/${_REPOSITORY}/${_IMAGE}:${_TAG}",
        f"serviceAccount: projects/{PROJECT}/serviceAccounts/{CLOUDBUILD_SERVICE_ACCOUNT}",
        "images:",
    ]
    absent = [item for item in required if item not in text]
    if absent:
        return Check("cloud_build_contract", "fail", f"missing_contract={absent}", current_tier="T1_model")

    return Check(
        "cloud_build_contract",
        "pass",
        "cloudbuild.gcp-staging.yaml builds, smoke-runs, then publishes Artifact Registry image",
        current_tier="T2_runtime",
    )


def check_cloud_build_service_account() -> Check:
    service_account = run_json(
        [
            "gcloud",
            "iam",
            "service-accounts",
            "describe",
            CLOUDBUILD_SERVICE_ACCOUNT,
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    policy = run_json(["gcloud", "projects", "get-iam-policy", PROJECT, "--format=json"])
    member = f"serviceAccount:{CLOUDBUILD_SERVICE_ACCOUNT}"
    roles = sorted(
        binding.get("role", "")
        for binding in policy.get("bindings", [])
        if member in binding.get("members", [])
    )
    missing = [role for role in CLOUDBUILD_REQUIRED_ROLES if role not in roles]
    if missing:
        return Check(
            "cloud_build_service_account",
            "fail",
            f"service_account={service_account.get('email')} missing_roles={missing}",
        )
    return Check(
        "cloud_build_service_account",
        "pass",
        f"service_account={service_account.get('email')} roles={','.join(roles)}",
    )


def check_github_actions_artifact_ci() -> Check:
    workflow = Path(".github/workflows/gcp-artifact.yml")
    if not workflow.exists():
        return Check(
            "github_actions_artifact_ci",
            "blocked",
            "missing .github/workflows/gcp-artifact.yml",
            required_tier="T3_live_readonly",
            current_tier="T0_spec",
        )

    text = workflow.read_text()
    required_text = [
        "google-github-actions/auth@v2",
        "workloadIdentityPools/github-actions/providers/living-ip-github",
        GITHUB_ARTIFACT_SERVICE_ACCOUNT,
        "docker run --rm",
        "docker push",
        "image_digest=",
        "image_ref=",
    ]
    absent = [item for item in required_text if item not in text]
    if absent:
        return Check("github_actions_artifact_ci", "fail", f"workflow_missing={absent}")

    project_number = run_text(["gcloud", "projects", "describe", PROJECT, "--format=value(projectNumber)"])
    provider = run_json(
        [
            "gcloud",
            "iam",
            "workload-identity-pools",
            "providers",
            "describe",
            GITHUB_WIF_PROVIDER,
            "--project",
            PROJECT,
            "--location",
            "global",
            "--workload-identity-pool",
            GITHUB_WIF_POOL,
            "--format=json",
        ]
    )
    condition = provider.get("attributeCondition", "")
    if GITHUB_REPOSITORY not in condition:
        return Check("github_actions_artifact_ci", "fail", f"provider_condition={condition!r}")

    service_account_policy = run_json(
        [
            "gcloud",
            "iam",
            "service-accounts",
            "get-iam-policy",
            GITHUB_ARTIFACT_SERVICE_ACCOUNT,
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    expected_member = (
        f"principalSet://iam.googleapis.com/projects/{project_number}/locations/global/"
        f"workloadIdentityPools/{GITHUB_WIF_POOL}/attribute.repository/{GITHUB_REPOSITORY}"
    )
    wif_bound = any(
        binding.get("role") == "roles/iam.workloadIdentityUser"
        and expected_member in binding.get("members", [])
        for binding in service_account_policy.get("bindings", [])
    )
    if not wif_bound:
        return Check("github_actions_artifact_ci", "fail", "missing_workloadIdentityUser_binding")

    project_policy = run_json(["gcloud", "projects", "get-iam-policy", PROJECT, "--format=json"])
    artifact_member = f"serviceAccount:{GITHUB_ARTIFACT_SERVICE_ACCOUNT}"
    artifact_writer = any(
        binding.get("role") == "roles/artifactregistry.writer"
        and artifact_member in binding.get("members", [])
        for binding in project_policy.get("bindings", [])
    )
    if not artifact_writer:
        return Check("github_actions_artifact_ci", "fail", "missing_artifactregistry_writer_role")

    return Check(
        "github_actions_artifact_ci",
        "pass",
        f"workflow=.github/workflows/gcp-artifact.yml provider={provider.get('name')} service_account={GITHUB_ARTIFACT_SERVICE_ACCOUNT}",
    )


def check_github_actions_readiness_ci() -> Check:
    workflow = Path(".github/workflows/gcp-readiness.yml")
    if not workflow.exists():
        return Check(
            "github_actions_readiness_ci",
            "blocked",
            "missing .github/workflows/gcp-readiness.yml",
            required_tier="T3_live_readonly",
            current_tier="T0_spec",
        )

    text = workflow.read_text()
    required = [
        "workflow_dispatch",
        "service_account:",
        "google-github-actions/auth@v2",
        "ops/check_gcp_infra_readiness.py",
        "READINESS_SERVICE_ACCOUNT",
        "READINESS_ARTIFACT_DIR: gcp-readiness-artifacts",
        "gcp-readiness-summary.json",
        "actions/upload-artifact@v4",
    ]
    absent = [item for item in required if item not in text]
    if absent:
        return Check(
            "github_actions_readiness_ci",
            "fail",
            f"workflow_missing={absent}",
            required_tier="T3_live_readonly",
            current_tier="T1_foundation",
        )
    return Check(
        "github_actions_readiness_ci",
        "pass",
        "workflow=.github/workflows/gcp-readiness.yml captures readiness stdout/stderr/exitcode through GitHub WIF in a non-hidden artifact directory",
        required_tier="T3_live_readonly",
        current_tier="T2_runtime",
    )


def _rule_allows_port(rule: dict, port: str) -> bool:
    for allowed in rule.get("allowed", []):
        if allowed.get("IPProtocol") in ("tcp", "all"):
            ports = allowed.get("ports")
            if not ports or port in ports:
                return True
    return False


def check_network_ingress() -> Check:
    rules = run_json(["gcloud", "compute", "firewall-rules", "list", "--project", PROJECT, "--format=json"])
    enabled_ingress = [
        rule
        for rule in rules
        if rule.get("direction", "INGRESS") == "INGRESS" and not rule.get("disabled", False)
    ]

    broad_sensitive = []
    for rule in enabled_ingress:
        sources = set(rule.get("sourceRanges", []))
        broad = bool({"0.0.0.0/0", "::/0"} & sources)
        if broad and (_rule_allows_port(rule, "22") or _rule_allows_port(rule, "3389")):
            broad_sensitive.append(rule.get("name", "unknown"))
    if broad_sensitive:
        return Check("network_ingress", "fail", f"broad_ssh_or_rdp_rules={broad_sensitive}")

    by_name = {rule.get("name"): rule for rule in rules}
    weak_ssh_rules = []
    for rule_name, target_tag in TELEO_SSH_RULES.items():
        rule = by_name.get(rule_name)
        if not rule or rule.get("disabled", False):
            weak_ssh_rules.append(f"{rule_name}:missing_or_disabled")
            continue
        sources = rule.get("sourceRanges", [])
        target_tags = rule.get("targetTags", [])
        if not sources or any(not source.endswith("/32") for source in sources):
            weak_ssh_rules.append(f"{rule_name}:sources={sources}")
        if target_tag not in target_tags:
            weak_ssh_rules.append(f"{rule_name}:targetTags={target_tags}")
        if not _rule_allows_port(rule, "22"):
            weak_ssh_rules.append(f"{rule_name}:does_not_allow_tcp_22")
    if weak_ssh_rules:
        return Check("network_ingress", "fail", " ".join(weak_ssh_rules))

    return Check(
        "network_ingress",
        "pass",
        "no enabled broad SSH/RDP ingress; Teleo SSH rules are /32-scoped and target-tagged",
    )


def check_compute_runtime_service_accounts() -> Check:
    instances = run_json(["gcloud", "compute", "instances", "list", "--project", PROJECT, "--format=json"])
    by_name = {instance.get("name"): instance for instance in instances}
    problems = []
    details = []
    for name, expected_email in TELEO_VMS.items():
        instance = by_name.get(name)
        if not instance:
            problems.append(f"{name}:missing")
            continue
        emails = [account.get("email") for account in instance.get("serviceAccounts", [])]
        details.append(f"{name}:{','.join(email for email in emails if email)}")
        if expected_email not in emails:
            problems.append(f"{name}:expected={expected_email}:actual={emails}")
        if any(email and email.endswith("-compute@developer.gserviceaccount.com") for email in emails):
            problems.append(f"{name}:uses_default_compute_service_account")
    if problems:
        return Check("compute_runtime_service_accounts", "fail", " ".join(problems))
    return Check("compute_runtime_service_accounts", "pass", " ".join(details))


def check_snapshot_policy() -> Check:
    policy = run_json(
        [
            "gcloud",
            "compute",
            "resource-policies",
            "describe",
            SNAPSHOT_POLICY,
            "--region",
            REGION,
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    disks = run_json(
        [
            "gcloud",
            "compute",
            "disks",
            "list",
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    policy_uri = policy["selfLink"]
    missing = [
        disk["name"]
        for disk in disks
        if disk["name"] in DISKS and policy_uri not in disk.get("resourcePolicies", [])
    ]
    if missing:
        return Check("compute_disk_snapshots", "fail", f"policy_not_attached={missing}")
    retention = policy.get("snapshotSchedulePolicy", {}).get("retentionPolicy", {})
    return Check(
        "compute_disk_snapshots",
        "pass",
        f"policy={SNAPSHOT_POLICY} attached_to={','.join(DISKS)} retention_days={retention.get('maxRetentionDays')}",
    )


def check_backup_buckets() -> Check:
    problems = []
    details = []
    for bucket in BACKUP_BUCKETS:
        versioning = run_text(["gsutil", "versioning", "get", f"gs://{bucket}"])
        ubla = run_text(["gsutil", "uniformbucketlevelaccess", "get", f"gs://{bucket}"])
        versioning_ok = "Enabled" in versioning
        ubla_ok = "Enabled: True" in ubla
        if not versioning_ok or not ubla_ok:
            problems.append(bucket)
        details.append(f"{bucket}:versioning={versioning_ok}:ubla={ubla_ok}")
    if problems:
        return Check("backup_buckets", "fail", " ".join(details))
    return Check("backup_buckets", "pass", " ".join(details))


def check_cloud_sql_standby() -> Check:
    instance = run_json(
        [
            "gcloud",
            "sql",
            "instances",
            "describe",
            CLOUDSQL_STANDBY_INSTANCE,
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    problems = []
    if instance.get("databaseVersion") != "POSTGRES_16":
        problems.append(f"databaseVersion={instance.get('databaseVersion')}")
    if instance.get("state") != "RUNNABLE":
        problems.append(f"state={instance.get('state')}")

    settings = instance.get("settings", {})
    if settings.get("deletionProtectionEnabled") is not True:
        problems.append("deletionProtectionEnabled=false")

    backup = settings.get("backupConfiguration", {})
    if backup.get("enabled") is not True:
        problems.append("backup_disabled")
    if backup.get("pointInTimeRecoveryEnabled") is not True:
        problems.append("pitr_disabled")
    retained = int(backup.get("backupRetentionSettings", {}).get("retainedBackups", 0) or 0)
    if retained < 7:
        problems.append(f"retainedBackups={retained}")
    log_days = int(backup.get("transactionLogRetentionDays", 0) or 0)
    if log_days < 7:
        problems.append(f"transactionLogRetentionDays={log_days}")

    ip_config = settings.get("ipConfiguration", {})
    if ip_config.get("ipv4Enabled") is not False:
        problems.append("public_ipv4_enabled")
    if ip_config.get("sslMode") != "ENCRYPTED_ONLY":
        problems.append(f"sslMode={ip_config.get('sslMode')}")
    if ip_config.get("privateNetwork") != f"projects/{PROJECT}/global/networks/teleo-staging-net":
        problems.append(f"privateNetwork={ip_config.get('privateNetwork')}")

    private_addresses = [
        address.get("ipAddress")
        for address in instance.get("ipAddresses", [])
        if address.get("type") == "PRIVATE"
    ]
    public_addresses = [
        address.get("ipAddress")
        for address in instance.get("ipAddresses", [])
        if address.get("type") in {"PRIMARY", "OUTGOING"}
    ]
    if not private_addresses:
        problems.append("missing_private_ip")
    if public_addresses:
        problems.append(f"public_addresses={public_addresses}")

    flags = {flag.get("name"): flag.get("value") for flag in settings.get("databaseFlags", [])}
    if flags.get("cloudsql.iam_authentication") != "on":
        problems.append("cloudsql.iam_authentication_not_on")

    databases = run_json(
        [
            "gcloud",
            "sql",
            "databases",
            "list",
            "--instance",
            CLOUDSQL_STANDBY_INSTANCE,
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    database_names = {database.get("name") for database in databases}
    if CLOUDSQL_STANDBY_DATABASE not in database_names:
        problems.append(f"missing_database={CLOUDSQL_STANDBY_DATABASE}")

    secret = run_json(
        [
            "gcloud",
            "secrets",
            "describe",
            CLOUDSQL_STANDBY_PASSWORD_SECRET,
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    versions = run_text(
        [
            "gcloud",
            "secrets",
            "versions",
            "list",
            CLOUDSQL_STANDBY_PASSWORD_SECRET,
            "--project",
            PROJECT,
            "--filter=state:ENABLED",
            "--format=value(name)",
        ]
    )
    if not versions:
        problems.append(f"secret_has_no_enabled_versions={secret.get('name')}")

    if problems:
        return Check("cloud_sql_standby_target", "fail", " ".join(problems))
    return Check(
        "cloud_sql_standby_target",
        "pass",
        f"instance={CLOUDSQL_STANDBY_INSTANCE} database={CLOUDSQL_STANDBY_DATABASE} private_ip={','.join(private_addresses)} backups=7 pitr=7d ssl=encrypted_only",
    )


def check_kb_source_restore_access() -> Check:
    secrets = run_json(["gcloud", "secrets", "list", "--project", PROJECT, "--format=json"])
    names = {secret["name"].split("/")[-1] for secret in secrets}
    source_candidates = sorted(
        name
        for name in names
        if any(token in name.lower() for token in ("source-kb", "kb-source", "source-postgres", "hetzner", "vps-db", "restore-source"))
    )
    if not source_candidates:
        return Check(
            "kb_source_restore_access",
            "blocked",
            f"no approved source KB/Postgres dump or replication credential found for restoring {KB_HOST} into GCP",
            required_tier="T4_restore_or_replication_readback",
            current_tier="T1_foundation",
        )
    return Check("kb_source_restore_access", "pass", "candidate_secret_names=" + ",".join(source_candidates))


def check_local_sqlite_postgres_restore_canary() -> Check:
    candidates = sorted(PROOF_ROOT.glob("sqlite-postgres-restore-canary-*.json"))
    if not candidates:
        return Check(
            "local_sqlite_postgres_restore_canary",
            "blocked",
            f"no sqlite-postgres restore canary proof found under {PROOF_ROOT}",
            required_tier="T2_local_restore_canary",
            current_tier="T1_foundation",
        )

    latest = candidates[-1]
    proof = json.loads(latest.read_text())
    mismatches = proof.get("mismatched_tables") or []
    source_tables = proof.get("source_table_count")
    target_tables = proof.get("target_table_count")
    source_rows = proof.get("source_total_rows")
    target_rows = proof.get("target_total_rows")
    if (
        proof.get("status") == "pass"
        and proof.get("source_integrity_check") == "ok"
        and not mismatches
        and source_tables == target_tables
        and source_rows == target_rows
    ):
        return Check(
            "local_sqlite_postgres_restore_canary",
            "pass",
            f"latest={latest} tables={source_tables} rows={source_rows} postgres_image={proof.get('postgres_image')}",
            required_tier="T2_local_restore_canary",
            current_tier="T2_local_restore_canary",
        )
    return Check(
        "local_sqlite_postgres_restore_canary",
        "fail",
        f"latest={latest} status={proof.get('status')} mismatches={len(mismatches)} source_tables={source_tables} target_tables={target_tables} source_rows={source_rows} target_rows={target_rows}",
        required_tier="T2_local_restore_canary",
        current_tier="T1_foundation",
    )


def check_gcp_cloudsql_restore_drill_contract() -> Check:
    script = Path("ops/run_gcp_cloudsql_restore_drill.sh")
    verifier = Path("ops/verify_gcp_cloudsql_restore_readback.py")
    runbook = Path("docs/gcp-kb-cloudsql-restore-runbook.md")
    missing = [str(path) for path in (script, verifier, runbook) if not path.exists()]
    if missing:
        return Check(
            "gcp_cloudsql_restore_drill_contract",
            "blocked",
            f"missing={missing}",
            required_tier="T2_restore_execution_contract",
            current_tier="T1_foundation",
        )
    script_text = script.read_text()
    runbook_text = runbook.read_text()
    required = [
        "gcloud storage cp",
        "gcloud sql import sql",
        "gcloud sql operations wait",
        "EXECUTE",
        "target-counts.sql",
        "source_table_counts",
    ]
    absent = [item for item in required if item not in script_text]
    verifier_text = verifier.read_text() if verifier.exists() else ""
    for item in ("target_counts_csv", "mismatched_tables", "source_total_rows", "target_total_rows"):
        if item not in verifier_text:
            absent.append(f"verifier:{item}")
    if "run_gcp_cloudsql_restore_drill.sh" not in runbook_text:
        absent.append("runbook_reference")
    if "verify_gcp_cloudsql_restore_readback.py" not in runbook_text:
        absent.append("readback_verifier_runbook_reference")
    if absent:
        return Check(
            "gcp_cloudsql_restore_drill_contract",
            "fail",
            f"missing_contract={absent}",
            required_tier="T2_restore_execution_contract",
            current_tier="T1_foundation",
        )
    return Check(
        "gcp_cloudsql_restore_drill_contract",
        "pass",
        "script can prepare GCS import SQL, optionally upload/import via Cloud SQL Admin, and emits target-count readback SQL",
        required_tier="T2_restore_execution_contract",
        current_tier="T2_restore_execution_contract",
    )


def check_kb_restore_or_replication() -> Check:
    return Check(
        "kb_restore_or_replication",
        "blocked",
        "GCP standby target exists, but no source-data restore, pg_dump import, logical replication, or query readback has been retained",
        required_tier="T4_restore_or_replication_readback",
        current_tier="T1_foundation",
    )


def main() -> int:
    checks = [
        safe_check("artifact_repositories", check_artifact_repositories),
        safe_check("cloud_build_contract", check_cloud_build_contract),
        safe_check("cloud_build_service_account", check_cloud_build_service_account),
        safe_check("github_actions_artifact_ci", check_github_actions_artifact_ci),
        safe_check("github_actions_readiness_ci", check_github_actions_readiness_ci),
        safe_check("network_ingress", check_network_ingress),
        safe_check("compute_runtime_service_accounts", check_compute_runtime_service_accounts),
        safe_check("compute_disk_snapshots", check_snapshot_policy),
        safe_check("backup_buckets", check_backup_buckets),
        safe_check("cloud_sql_standby_target", check_cloud_sql_standby),
        safe_check("kb_source_restore_access", check_kb_source_restore_access),
        safe_check("local_sqlite_postgres_restore_canary", check_local_sqlite_postgres_restore_canary),
        safe_check("gcp_cloudsql_restore_drill_contract", check_gcp_cloudsql_restore_drill_contract),
        safe_check("kb_restore_or_replication", check_kb_restore_or_replication),
    ]
    payload = {
        "project": PROJECT,
        "region": REGION,
        "zone": ZONE,
        "checks": [check.__dict__ for check in checks],
        "pass_count": sum(check.status == "pass" for check in checks),
        "blocked_count": sum(check.status == "blocked" for check in checks),
        "fail_count": sum(check.status == "fail" for check in checks),
    }
    print(json.dumps(payload, indent=2))
    return 1 if payload["fail_count"] else 0


if __name__ == "__main__":
    raise SystemExit(main())