Check GCP communication posture (#40)

docs/gcp-infra-hardening-20260707.md

@@ -70,6 +70,10 @@ The check is read-only and prints no secret values. It verifies:
 - Cloud Build config contract.
 - Dedicated Cloud Build service account and roles.
 - GitHub Actions WIF Artifact Registry publishing contract.
+- Network ingress posture:
+  - no enabled broad SSH/RDP ingress;
+  - Teleo SSH rules are scoped to `/32` source ranges and target tags.
+- Runtime service-account posture for the prod/staging VMs.
 - Compute disk snapshot policy attachment.
 - Backup bucket versioning and uniform access.
 - Whether Cloud SQL/Postgres exists.
@@ -94,5 +98,6 @@ 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.
+- prod/staging VMs run as dedicated Teleo service accounts, not the default Compute Engine service account.
 
 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

@@ -22,6 +22,14 @@ 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",
@@ -241,6 +249,78 @@ def check_github_actions_artifact_ci() -> Check:
     )
 
 
+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(
         [
@@ -344,6 +424,8 @@ def main() -> int:
         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("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_pgvector", check_cloud_sql),