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 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"
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"


@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",
    ]
    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_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() -> Check:
    instances = run_json(
        [
            "gcloud",
            "sql",
            "instances",
            "list",
            "--project",
            PROJECT,
            "--format=json",
        ]
    )
    if not instances:
        return Check(
            "cloud_sql_pgvector",
            "blocked",
            "no Cloud SQL/Postgres instance exists in GCP; canonical KB still lives outside this GCP project",
            required_tier="T3_live_readonly",
            current_tier="T0_spec",
        )
    names = ",".join(instance["name"] for instance in instances)
    return Check("cloud_sql_pgvector", "pass", f"instances={names}")


def check_kb_credential_slot() -> Check:
    secrets = run_json(["gcloud", "secrets", "list", "--project", PROJECT, "--format=json"])
    names = {secret["name"].split("/")[-1] for secret in secrets}
    candidates = sorted(name for name in names if any(token in name.lower() for token in ("kb", "postgres", "pg", "ssh")))
    if not candidates:
        return Check(
            "kb_gcp_access_secret",
            "blocked",
            f"no KB/Postgres/SSH secret slot found for approved GCP-to-KB access; current blocker remains {KB_HOST} auth",
            required_tier="T3_live_readonly",
            current_tier="T0_spec",
        )
    return Check("kb_gcp_access_secret", "pass", "candidate_secret_names=" + ",".join(candidates))


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("compute_disk_snapshots", check_snapshot_policy),
        safe_check("backup_buckets", check_backup_buckets),
        safe_check("cloud_sql_pgvector", check_cloud_sql),
        safe_check("kb_gcp_access_secret", check_kb_credential_slot),
    ]
    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())