teleo-infrastructure/ops/check_gcp_service_communications.py

#!/usr/bin/env python3
"""Validate the Teleo GCP service-to-service communication contract."""

from __future__ import annotations

import argparse
import ipaddress
import json
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

BROAD_CIDRS = {"0.0.0.0/0", "::/0"}
SECRET_FIELD_TOKENS = ("password", "private_key", "secret_value", "token_value", "seed_phrase")
DEFAULT_COMPUTE_SUFFIX = "-compute@developer.gserviceaccount.com"
REQUIRED_PATHS = {
    "github_actions_to_artifact_registry",
    "github_actions_to_readiness_probe",
    "operator_ssh_to_teleo_vms",
    "teleo_vms_to_artifact_registry",
    "teleo_vms_to_cloudsql_private",
    "cloudsql_import_from_backup_bucket",
    "restore_drill_operator_to_cloudsql_admin",
    "vps_backup_to_gcs_bucket",
}


def load_contract(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text())


def as_list(value: Any) -> list[Any]:
    return value if isinstance(value, list) else []


def nested_values(value: Any) -> list[Any]:
    if isinstance(value, dict):
        result: list[Any] = []
        for item in value.values():
            result.extend(nested_values(item))
        return result
    if isinstance(value, list):
        result = []
        for item in value:
            result.extend(nested_values(item))
        return result
    return [value]


def find_secret_fields(value: Any, prefix: str = "") -> list[str]:
    problems: list[str] = []
    if isinstance(value, dict):
        for key, item in value.items():
            path = f"{prefix}.{key}" if prefix else str(key)
            if str(key).lower() in SECRET_FIELD_TOKENS:
                problems.append(path)
            problems.extend(find_secret_fields(item, path))
    elif isinstance(value, list):
        for index, item in enumerate(value):
            problems.extend(find_secret_fields(item, f"{prefix}[{index}]"))
    return problems


def validate_cidrs(path: dict[str, Any]) -> list[str]:
    problems: list[str] = []
    cidrs = as_list(path.get("source", {}).get("cidrs"))
    for cidr in cidrs:
        if cidr in BROAD_CIDRS:
            problems.append(f"{path['name']}:broad_source_cidr={cidr}")
            continue
        if cidr == "<operator-ip>/32":
            continue
        try:
            network = ipaddress.ip_network(str(cidr), strict=False)
        except ValueError:
            problems.append(f"{path['name']}:invalid_source_cidr={cidr}")
            continue
        if path.get("ports") == [22] and (network.version != 4 or network.prefixlen != 32):
            problems.append(f"{path['name']}:ssh_source_not_single_ipv4_32={cidr}")
    return problems


def validate_path(path: dict[str, Any]) -> list[str]:
    problems: list[str] = []
    name = path.get("name", "<unnamed>")
    for field in ("name", "purpose", "source", "identity", "target", "protocol", "ports", "network_path", "encryption", "allowed_by"):
        if field not in path:
            problems.append(f"{name}:missing_field={field}")

    ports = as_list(path.get("ports"))
    if any(not isinstance(port, int) for port in ports):
        problems.append(f"{name}:ports_must_be_integers")
    if 22 in ports and name != "operator_ssh_to_teleo_vms":
        problems.append(f"{name}:tcp_22_only_allowed_for_operator_ssh")
    if 3389 in ports:
        problems.append(f"{name}:rdp_forbidden")

    problems.extend(validate_cidrs(path))

    values = [str(value) for value in nested_values(path)]
    if any(value in BROAD_CIDRS for value in values):
        problems.append(f"{name}:contains_broad_cidr")
    if any(value.endswith(DEFAULT_COMPUTE_SUFFIX) for value in values):
        problems.append(f"{name}:default_compute_service_account_forbidden")

    target = path.get("target", {})
    if target.get("type") == "cloudsql_postgres":
        if target.get("public_ip") is not False:
            problems.append(f"{name}:cloudsql_public_ip_not_false")
        if not target.get("private_network"):
            problems.append(f"{name}:cloudsql_missing_private_network")
        if path.get("encryption") != "encrypted_only":
            problems.append(f"{name}:cloudsql_encryption_not_encrypted_only")

    if path.get("protocol") in {"https", "postgres"} and path.get("encryption") in {None, "none", "disabled"}:
        problems.append(f"{name}:encrypted_protocol_without_encryption")
    return problems


def build_payload(contract_path: Path) -> dict[str, Any]:
    contract = load_contract(contract_path)
    paths = as_list(contract.get("allowed_paths"))
    names = {str(path.get("name")) for path in paths if isinstance(path, dict)}
    problems: list[str] = []

    missing_paths = sorted(REQUIRED_PATHS - names)
    if missing_paths:
        problems.append(f"missing_required_paths={missing_paths}")
    extra_secret_fields = find_secret_fields(contract)
    if extra_secret_fields:
        problems.append(f"secret_value_fields_present={extra_secret_fields}")

    invariants = contract.get("global_invariants", {})
    for invariant in (
        "no_public_database_ip",
        "no_broad_ssh_or_rdp",
        "no_default_compute_service_accounts",
        "secret_values_not_stored_in_contract",
        "database_connections_encrypted_only",
    ):
        if invariants.get(invariant) is not True:
            problems.append(f"invariant_not_true={invariant}")

    for path in paths:
        if not isinstance(path, dict):
            problems.append("allowed_paths_contains_non_object")
            continue
        problems.extend(validate_path(path))

    return {
        "artifact": "teleo_gcp_service_communications_check",
        "generated_at_utc": datetime.now(UTC).isoformat(),
        "contract_path": str(contract_path),
        "project": contract.get("project"),
        "network": contract.get("network"),
        "allowed_path_count": len(paths),
        "required_path_count": len(REQUIRED_PATHS),
        "status": "pass" if not problems else "fail",
        "problems": problems,
        "not_proven_by_this_artifact": contract.get("not_proven_by_this_contract", []),
    }


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--contract", default="config/gcp-service-communications.json", type=Path)
    parser.add_argument(
        "--output",
        default="outputs/gcp-infra-hardening-20260707/proofs/gcp-service-communications-check.json",
        type=Path,
    )
    args = parser.parse_args()
    payload = build_payload(args.contract)
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n")
    print(json.dumps(payload, indent=2, sort_keys=True))
    return 0 if payload["status"] == "pass" else 1


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