Harden GCP Docker build and infra readiness (#38)

cloudbuild.gcp-staging.yaml

@@ -1,11 +1,43 @@
+substitutions:
+  _REGION: europe-west6
+  _REPOSITORY: teleo
+  _IMAGE: teleo-pipeline-gcp-staging
+  _TAG: manual-local
+  _REVISION: unknown
+
+options:
+  logging: CLOUD_LOGGING_ONLY
+  machineType: E2_HIGHCPU_8
+
+serviceAccount: projects/teleo-501523/serviceAccounts/sa-teleo-cloudbuild@teleo-501523.iam.gserviceaccount.com
+
 steps:
-  - name: gcr.io/cloud-builders/docker
+  - id: build-staging-image
+    name: gcr.io/cloud-builders/docker
     args:
       - build
       - -f
       - Dockerfile.gcp-staging
+      - --label
+      - org.opencontainers.image.source=https://github.com/living-ip/teleo-infrastructure
+      - --label
+      - org.opencontainers.image.revision=${_REVISION}
+      - --label
+      - livingip.revision=${_REVISION}
+      - --label
+      - livingip.surface=teleo-infrastructure
+      - --label
+      - livingip.tier=gcp-staging
       - -t
-      - europe-west6-docker.pkg.dev/teleo-501523/teleo/teleo-pipeline-gcp-staging:66ecbf3-20260706
+      - ${_REGION}-docker.pkg.dev/$PROJECT_ID/${_REPOSITORY}/${_IMAGE}:${_TAG}
       - .
+
+  - id: smoke-test-image-before-push
+    name: gcr.io/cloud-builders/docker
+    args:
+      - run
+      - --rm
+      - ${_REGION}-docker.pkg.dev/$PROJECT_ID/${_REPOSITORY}/${_IMAGE}:${_TAG}
+
 images:
-  - europe-west6-docker.pkg.dev/teleo-501523/teleo/teleo-pipeline-gcp-staging:66ecbf3-20260706
+  - ${_REGION}-docker.pkg.dev/$PROJECT_ID/${_REPOSITORY}/${_IMAGE}:${_TAG}

docs/gcp-infra-hardening-20260707.md

@@ -0,0 +1,82 @@
+# GCP CI/CD and Redundancy Hardening
+
+This records the current GCP hardening contract for `teleo-501523`.
+
+## What Is Live
+
+- Artifact Registry Docker repositories exist in `europe-west6`:
+  - `teleo`
+  - `livingip-web`
+- Both repositories use immutable tags and active vulnerability scanning.
+- `cloudbuild.gcp-staging.yaml` builds the staging Teleo image, runs the image smoke test, then pushes to Artifact Registry.
+- Cloud Build runs as the dedicated service account:
+  - `sa-teleo-cloudbuild@teleo-501523.iam.gserviceaccount.com`
+- The dedicated Cloud Build account has only the roles required for the current build/publish path:
+  - `roles/artifactregistry.writer`
+  - `roles/logging.logWriter`
+  - `roles/storage.objectViewer`
+- Backup buckets are versioned and use uniform bucket-level access:
+  - `gs://teleo-501523-prod-backups`
+  - `gs://teleo-501523-leoclean-backups`
+- VM boot disks have a daily 7-day snapshot policy:
+  - `teleo-prod-1`
+  - `teleo-staging-1`
+
+## How To Build
+
+```bash
+REVISION="$(git rev-parse HEAD)"
+TAG="$(git rev-parse --short=7 HEAD)-manual-$(date -u +%Y%m%d-%H%M%S)"
+
+gcloud builds submit \
+  --project=teleo-501523 \
+  --config=cloudbuild.gcp-staging.yaml \
+  --substitutions="_TAG=${TAG},_REVISION=${REVISION}"
+```
+
+Expected result:
+
+- `build-staging-image` succeeds.
+- `smoke-test-image-before-push` succeeds.
+- A Docker image is pushed to:
+  `europe-west6-docker.pkg.dev/teleo-501523/teleo/teleo-pipeline-gcp-staging:${TAG}`
+
+## How To Check Readiness
+
+Run from the repository root:
+
+```bash
+python3 ops/check_gcp_infra_readiness.py
+```
+
+The check is read-only and prints no secret values. It verifies:
+
+- Artifact Registry immutability and vulnerability scanning.
+- Cloud Build config contract.
+- Dedicated Cloud Build service account and roles.
+- Compute disk snapshot policy attachment.
+- Backup bucket versioning and uniform access.
+- Whether Cloud SQL/Postgres exists.
+- Whether a GCP-approved KB/Postgres access secret slot exists.
+
+## Current Boundaries
+
+The GCP Docker build/publish path is live and proved manually. Automatic Git-triggered CI/CD is not yet live because this project currently has no Cloud Build repository connection or trigger.
+
+Database redundancy is not complete. The current project has backup buckets and VM disk snapshots, but it does not yet have a GCP Cloud SQL/Postgres pgvector standby or primary. Do not claim DB parity until one of these is true:
+
+- the existing canonical KB database is replicated into GCP and read back; or
+- GCP Cloud SQL/Postgres becomes the canonical database and production services read/write it; or
+- an explicitly approved standby restore drill proves that a GCP database can be restored and queried from the retained backups.
+
+Do not create an empty Cloud SQL instance and call it redundancy. It only counts after source data, restore/replication, access controls, and readback are proven.
+
+## Communication Posture
+
+Current GCP communication hardening is limited to the surfaces already read back:
+
+- default SSH/RDP firewall rules are disabled;
+- Teleo SSH access is scoped by instance tags and current source IP rules;
+- the custom Teleo subnet has Private Google Access enabled.
+
+Before production cutover, add a dedicated runtime communication contract for service-to-service paths, including exact source tags/service accounts, allowed ports, health checks, and rollback.

ops/check_gcp_infra_readiness.py

@@ -0,0 +1,275 @@
+#!/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",
+)
+
+
+@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_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("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())